This is an automated email from the ASF dual-hosted git repository.
lgbo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 899f7d81d9 [GLUTEN-9812][Flink] Support type Date (#9813)
899f7d81d9 is described below
commit 899f7d81d9e66e6b58f42529161b543e731c4d03
Author: lgbo <[email protected]>
AuthorDate: Fri Jun 6 09:38:22 2025 +0800
[GLUTEN-9812][Flink] Support type Date (#9813)
---
.../apache/gluten/util/LogicalTypeConverter.java | 3 +++
.../gluten/vectorized/ArrowVectorAccessor.java | 15 ++++++++++-
.../gluten/vectorized/ArrowVectorWriter.java | 30 +++++++++++++++++++++-
.../table/runtime/stream/custom/ScanTest.java | 10 ++++++++
4 files changed, 56 insertions(+), 2 deletions(-)
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
index b71f906e18..7db368ae14 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BooleanType;
import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DateType;
import org.apache.flink.table.types.logical.DayTimeIntervalType;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.DoubleType;
@@ -75,6 +76,8 @@ public class LogicalTypeConverter {
Map.entry(
DayTimeIntervalType.class,
logicalType -> new
io.github.zhztheplayer.velox4j.type.BigIntType()),
+ Map.entry(
+ DateType.class, logicalType -> new
io.github.zhztheplayer.velox4j.type.DateType()),
Map.entry(
RowType.class,
logicalType -> {
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
index 765c24179c..eecdf957da 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.arrow.vector.BigIntVector;
import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.Float8Vector;
import org.apache.arrow.vector.IntVector;
@@ -57,6 +58,7 @@ public abstract class ArrowVectorAccessor {
Map.entry(VarCharVector.class, vector -> new
VarCharVectorAccessor(vector)),
Map.entry(StructVector.class, vector -> new
StructVectorAccessor(vector)),
Map.entry(ListVector.class, vector -> new
ListVectorAccessor(vector)),
+ Map.entry(DateDayVector.class, vector -> new
DateDayVectorAccessor(vector)),
Map.entry(MapVector.class, vector -> new MapVectorAccessor(vector)));
public static ArrowVectorAccessor create(FieldVector vector) {
@@ -145,8 +147,19 @@ class DoubleVectorAccessor extends
BaseArrowVectorAccessor<Float8Vector> {
}
}
-class VarCharVectorAccessor extends BaseArrowVectorAccessor<VarCharVector> {
+class DateDayVectorAccessor extends BaseArrowVectorAccessor<DateDayVector> {
+
+ public DateDayVectorAccessor(FieldVector vector) {
+ super(vector);
+ }
+ @Override
+ protected Object getImpl(int rowIndex) {
+ return typedVector.get(rowIndex);
+ }
+}
+
+class VarCharVectorAccessor extends BaseArrowVectorAccessor<VarCharVector> {
public VarCharVectorAccessor(FieldVector vector) {
super(vector);
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
index 88b822e435..80d3bd4b7a 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
@@ -24,9 +24,11 @@ import org.apache.flink.table.data.RowData;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.*;
+import org.apache.arrow.vector.DateDayVector;
import org.apache.arrow.vector.complex.ListVector;
import org.apache.arrow.vector.complex.MapVector;
import org.apache.arrow.vector.complex.StructVector;
+import org.apache.arrow.vector.types.DateUnit;
import org.apache.arrow.vector.types.FloatingPointPrecision;
import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.pojo.*;
@@ -67,6 +69,10 @@ public abstract class ArrowVectorWriter {
TimestampType.class,
(fieldType, allocator, vector) ->
new TimestampVectorWriter(fieldType, allocator, vector)),
+ Map.entry(
+ DateType.class,
+ (fieldType, allocator, vector) ->
+ new DateDayVectorWriter(fieldType, allocator, vector)),
Map.entry(
RowType.class,
(fieldType, allocator, vector) ->
@@ -152,7 +158,8 @@ class FieldVectorCreator {
TimestampType.class,
(dataType, timeZoneId) ->
new ArrowType.Timestamp(
- TimeUnit.MILLISECOND, timeZoneId == null ? "UTC" :
timeZoneId)));
+ TimeUnit.MILLISECOND, timeZoneId == null ? "UTC" :
timeZoneId)),
+ Map.entry(DateType.class, (dataType, timeZoneId) -> new
ArrowType.Date(DateUnit.DAY)));
private static ArrowType toArrowType(Type dataType, String timeZoneId) {
ArrowTypeConverter converter =
arrowTypeConverters.get(dataType.getClass());
@@ -375,6 +382,27 @@ class TimestampVectorWriter extends
BaseVectorWriter<TimeStampMilliVector, Long>
}
}
+class DateDayVectorWriter extends BaseVectorWriter<DateDayVector, Integer> {
+ public DateDayVectorWriter(Type fieldType, BufferAllocator allocator,
FieldVector vector) {
+ super(vector);
+ }
+
+ @Override
+ protected Integer getValue(RowData rowData, int fieldIndex) {
+ return rowData.getInt(fieldIndex);
+ }
+
+ @Override
+ protected Integer getValue(ArrayData arrayData, int index) {
+ return arrayData.getInt(index);
+ }
+
+ @Override
+ protected void setValue(int index, Integer value) {
+ this.typedVector.setSafe(index, value);
+ }
+}
+
class StructVectorWriter extends BaseVectorWriter<StructVector, RowData> {
private final int fieldCount;
private BufferAllocator allocator;
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScanTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScanTest.java
index d50a057959..23fcc46db9 100644
---
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScanTest.java
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScanTest.java
@@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.LocalDate;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
@@ -177,4 +178,13 @@ class ScanTest extends GlutenStreamingTestBase {
String query = "select a, b from charTbl where a > 0";
runAndCheck(query, Arrays.asList("+I[1, a1]", "+I[2, b2]", "+I[3, c2]"));
}
+
+ void testDateScan() {
+ List<Row> rows =
+ Arrays.asList(
+ Row.of(1, LocalDate.parse("2023-01-01")), Row.of(2,
LocalDate.parse("2023-01-02")));
+ createSimpleBoundedValuesTable("dateTbl", "a int, b date", rows);
+ String query = "select a, b from dateTbl where a > 0";
+ runAndCheck(query, Arrays.asList("+I[1, 2023-01-01]", "+I[2,
2023-01-02]"));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
