Skip to content
Open
Show file tree
Hide file tree
Changes from 12 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/flink.yml
Original file line number Diff line number Diff line change
Expand Up @@ -89,9 +89,9 @@ jobs:
export VELOX_DEPENDENCY_SOURCE=BUNDLED
export fmt_SOURCE=BUNDLED
export folly_SOURCE=BUNDLED
git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
git clone -b gluten-20260829 https://github.com/TianJialong/velox4j.git
Comment thread
TianJialong marked this conversation as resolved.
Resolved
cd velox4j
git reset --hard $(cat $GITHUB_WORKSPACE/gluten-flink/velox4j-ref.txt)
# use latest commit
git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
$GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
cd ..
Expand Down
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
*.iml
.DS_Store
*.log
*.out
Comment thread
TianJialong marked this conversation as resolved.
Resolved
**/core$
hs_err_pid*
*.cache
Expand Down
4 changes: 2 additions & 2 deletions gluten-flink/docs/Flink.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,9 +50,9 @@ As some features have not been committed to upstream, you have to use the follow

```bash
## fetch velox4j code
git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
git clone -b gluten-20260829 https://github.com/TianJialong/velox4j.git
cd velox4j
git reset --hard $(cat $GLUTEN_FLINK_HOME/velox4j-ref.txt)
# use latest commit
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,11 +43,13 @@
import org.apache.flink.util.Preconditions;

import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexFieldAccess;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.util.NlsString;

Expand All @@ -67,6 +69,14 @@ public static TypedExpr toTypedExpr(RexNode rexNode, RexConversionContext contex
return new ConstantTypedExpr(toType(literal.getType()), toVariant(literal), null);
} else if (rexNode instanceof RexCall) {
RexCall rexCall = (RexCall) rexNode;
if (rexCall.getOperator().getKind() == SqlKind.CAST && rexCall.getOperands().size() == 1) {
RexNode operand = rexCall.getOperands().get(0);
// An identity cast (types equal ignoring only nullability, e.g. around folded
// literals) carries no semantics; velox rejects same-type casts, so pass through.
if (isSameTypeIgnoringNullability(rexCall.getType(), operand.getType())) {
return toTypedExpr(operand, context);
}
}
RexCallConverter converter = RexCallConverterFactory.getConverter(rexCall, context);
return converter.toTypedExpr(rexCall, context);
} else if (rexNode instanceof RexInputRef) {
Expand Down Expand Up @@ -97,6 +107,48 @@ public static Type toType(RelDataType relDataType) {
return LogicalTypeConverter.toVLType(FlinkTypeFactory.toLogicalType(relDataType));
}

/**
* Structural equality ignoring nullability only: name, precision, scale, and every component type
* recursively (field names included). ANY is opaque and never qualifies.
*/
private static boolean isSameTypeIgnoringNullability(RelDataType a, RelDataType b) {
if (a == b) {
return true;
}
SqlTypeName kind = a.getSqlTypeName();
if (kind != b.getSqlTypeName()
|| a.getPrecision() != b.getPrecision()
|| a.getScale() != b.getScale()) {
return false;
}
if (kind == SqlTypeName.ARRAY || kind == SqlTypeName.MULTISET) {
return isSameTypeIgnoringNullability(a.getComponentType(), b.getComponentType());
}
if (kind == SqlTypeName.MAP) {
return isSameTypeIgnoringNullability(a.getKeyType(), b.getKeyType())
&& isSameTypeIgnoringNullability(a.getValueType(), b.getValueType());
}
if (kind == SqlTypeName.ANY) {
return false;
}
if (a.isStruct()) {
if (a.getFieldCount() != b.getFieldCount()) {
return false;
}
for (int i = 0; i < a.getFieldCount(); i++) {
RelDataTypeField fieldA = a.getFieldList().get(i);
RelDataTypeField fieldB = b.getFieldList().get(i);
if (!fieldA.getName().equals(fieldB.getName())
|| !isSameTypeIgnoringNullability(fieldA.getType(), fieldB.getType())) {
return false;
}
}
return true;
}
// Scalars: name + precision + scale carry the identity (charset/collation are factory-wide).
return true;
}

private static Variant toNullVariant(SqlTypeName sqlTypeName) {
// A typed null needs no per-type carrier: the velox-side
// ConstantTypedExpr renders a null variant as a null constant of the
Expand All @@ -121,6 +173,7 @@ private static Variant toNullVariant(SqlTypeName sqlTypeName) {
case SYMBOL:
case TIME:
case TIMESTAMP:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return new VarCharValue(null);
default:
throw new RuntimeException("Unsupported null rex node type: " + sqlTypeName);
Expand Down Expand Up @@ -168,13 +221,27 @@ public static Variant toVariant(RexLiteral literal) {
}
case SYMBOL:
return new VarCharValue(literal.getValue().toString());
case DATE:
// Calcite keeps a DATE literal as int epoch days — the velox DATE representation; NULL
// maps to a null variant (folded NULLs of any type still fail velox4j serialization).
Integer dateValue = literal.getValueAs(Integer.class);
return dateValue == null ? new VarCharValue(null) : new IntegerValue(dateValue);
case TIMESTAMP:
// RexLiteral#getValueAs does not support java.sql.Timestamp, so take the
// epoch millis directly. It keeps the wall time as UTC, which is the same
// base as TimestampData#getMillisecond used by the row serializer.
long epochMillis = literal.getValueAs(Long.class);
return TimestampValue.create(
Math.floorDiv(epochMillis, 1000L), Math.floorMod(epochMillis, 1000L) * 1_000_000L);
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
// Calcite 1.32 resolves both names through the same branch
// (TimestampString.getMillisSinceEpoch()); Flink's folding round-trip
// stores an LTZ literal as a GMT-frame TimestampString and reads it
// back atOffset(UTC) — the same epoch-millis base as TIMESTAMP.
// RexLiteral#getValueAs does not support java.sql.Timestamp, so take
// the epoch millis directly. It keeps the wall time as UTC, which is
// the same base as TimestampData#getMillisecond used by the row
// serializer. A NULL literal maps to a null variant (planner no
// longer NPEs; velox4j serialization of inner nulls is a known gap).
Long epochMillis = literal.getValueAs(Long.class);
return epochMillis == null
? new VarCharValue(null)
: TimestampValue.create(
Math.floorDiv(epochMillis, 1000L), Math.floorMod(epochMillis, 1000L) * 1_000_000L);
case TIME:
// Calcite stores TIME literals as the millisecond-of-day Integer —
// the same encoding the velox side uses for TIME.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,18 @@ public class RexCallConverterFactory {
Map.entry("IS NULL", Arrays.asList(() -> new DefaultRexCallConverter("isnull"))),
Map.entry(
"REGEXP_EXTRACT", Arrays.asList(() -> new DefaultRexCallConverter("regexp_extract"))),
// Direct name mapping: both arities are covered by the flinksql to_date signatures.
Map.entry("TO_DATE", Arrays.asList(() -> new DefaultRexCallConverter("to_date"))),
Comment thread
TianJialong marked this conversation as resolved.
Resolved
// Direct name mapping: both arities are covered by the flinksql to_timestamp
// signatures; the returned TIMESTAMP(3) maps to the velox TimestampType.
Map.entry(
"TO_TIMESTAMP", Arrays.asList(() -> new DefaultRexCallConverter("to_timestamp"))),
// Direct name mapping: the only SQL form is (numeric, precision) — the first
// operand's type is routed by the velox signature set, the returned
// TIMESTAMP_LTZ(3) maps to the velox TimestampType.
Map.entry(
"TO_TIMESTAMP_LTZ",
Arrays.asList(() -> new DefaultRexCallConverter("to_timestamp_ltz"))),
Map.entry("LOWER", Arrays.asList(() -> new DefaultRexCallConverter("lower"))),
Map.entry("count_char", Arrays.asList(() -> new DefaultRexCallConverter("count_char"))),
Map.entry("EXTRACT", Arrays.asList(() -> new DefaultRexCallConverter("extract"))),
Expand Down
Loading
Loading