Skip to content

Commit abeeff4

Browse files
fresh-borzonifresh-borzoni
committed
address comments
1 parent d373e0d commit abeeff4

7 files changed

Lines changed: 58 additions & 57 deletions

File tree

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java‎

Lines changed: 2 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1228,15 +1228,15 @@ void testStreamingReadAllPartitionTypePushDown() throws Exception {
12281228

12291229
String query =
12301230
"select * from all_type_partitioned_table where "
1231-
+ "p_bool is false and p_int not in (12) and p_bigint in (99999) and p_bytes=CAST('Hi' AS VARBINARY) "
1231+
+ "p_bool is false and p_int < 11 and p_bigint in (99999) and p_bytes=CAST('Hi' AS VARBINARY) "
12321232
+ "and p_string='hello' and p_float=CAST(12.5 AS FLOAT) and p_double=7.88 and p_date=DATE '2025-10-12' "
12331233
+ "and p_time=TIME '12:55:00' and p_ts_ntz=TIMESTAMP '2025-10-12 12:55:00.001' "
12341234
+ "and p_ts_ltz=TO_TIMESTAMP_LTZ(4001, 3)";
12351235
String plan = tEnv.explainSql(query);
12361236
assertThat(plan)
12371237
.contains(
12381238
"TableSourceScan(table=[[testcatalog, defaultdb, all_type_partitioned_table, "
1239-
+ "filter=[and(and(and(and(and(and(and(and(and(and(<>(p_int, 12), "
1239+
+ "filter=[and(and(and(and(and(and(and(and(and(and(<(p_int, 11), "
12401240
+ "=(p_bigint, 99999:BIGINT)), =(p_bytes, X'4869':VARBINARY(2147483647))), "
12411241
+ "=(p_string, _UTF-16LE'hello':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\")), "
12421242
+ "=(p_float, 1.25E1:FLOAT)), =(p_double, 7.88E0:DOUBLE)), =(p_date, 2025-10-12)), "
@@ -1756,32 +1756,6 @@ private List<String> writeRowsToTwoPartition(TablePath tablePath, Collection<Str
17561756
return expectedRowValues;
17571757
}
17581758

1759-
@Test
1760-
void testReadPartitionPushDownWithIntPartitionRangePredicate() throws Exception {
1761-
tEnv.executeSql(
1762-
"create table int_partitioned_table"
1763-
+ " (a int not null, b varchar, pt int, primary key (a, pt) NOT ENFORCED) partitioned by (pt) ");
1764-
TablePath tablePath = TablePath.of(DEFAULT_DB, "int_partitioned_table");
1765-
tEnv.executeSql("alter table int_partitioned_table add partition (pt=2)");
1766-
tEnv.executeSql("alter table int_partitioned_table add partition (pt=10)");
1767-
1768-
List<InternalRow> rows = new ArrayList<>();
1769-
for (int i = 0; i < 3; i++) {
1770-
rows.add(row(i, "v" + i, 2));
1771-
rows.add(row(i, "v" + i, 10));
1772-
}
1773-
writeRows(conn, tablePath, rows, false);
1774-
FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath);
1775-
1776-
List<String> expected = new ArrayList<>();
1777-
for (int i = 0; i < 3; i++) {
1778-
expected.add(String.format("+I[%d, v%d, 10]", i, i));
1779-
}
1780-
CloseableIterator<Row> rowIter =
1781-
tEnv.executeSql("select * from int_partitioned_table where pt > 2").collect();
1782-
assertResultsIgnoreOrder(rowIter, expected, true);
1783-
}
1784-
17851759
@Test
17861760
void testStreamingReadPartitionPushDownWithInExpr() throws Exception {
17871761

‎fluss-rpc/src/main/java/org/apache/fluss/rpc/util/PredicateMessageUtils.java‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,6 @@
5151
import org.apache.fluss.types.DecimalType;
5252
import org.apache.fluss.types.RowType;
5353

54-
import java.time.LocalDate;
55-
import java.time.LocalTime;
5654
import java.util.ArrayList;
5755
import java.util.HashMap;
5856
import java.util.List;
@@ -145,9 +143,9 @@ private static Object toLiteralValue(PbLiteralValue pbLiteral, DataType fieldTyp
145143
decimalType.getScale());
146144
}
147145
case DATE:
148-
return LocalDate.ofEpochDay(pbLiteral.getBigintValue());
146+
return (int) pbLiteral.getBigintValue();
149147
case TIME_WITHOUT_TIME_ZONE:
150-
return LocalTime.ofNanoOfDay(pbLiteral.getIntValue() * 1_000_000L);
148+
return pbLiteral.getIntValue();
151149
case TIMESTAMP_WITHOUT_TIME_ZONE:
152150
return TimestampNtz.fromMillis(
153151
pbLiteral.getTimestampMillisValue(),
@@ -282,10 +280,10 @@ private static PbLiteralValue toPbLiteralValue(DataType type, Object literal) {
282280
pbLiteral.setIntValue((Integer) literal);
283281
break;
284282
case DATE:
285-
pbLiteral.setBigintValue(((LocalDate) literal).toEpochDay());
283+
pbLiteral.setBigintValue(((Integer) literal).longValue());
286284
break;
287285
case TIME_WITHOUT_TIME_ZONE:
288-
pbLiteral.setIntValue((int) (((LocalTime) literal).toNanoOfDay() / 1_000_000L));
286+
pbLiteral.setIntValue((Integer) literal);
289287
break;
290288
case TIMESTAMP_WITHOUT_TIME_ZONE:
291289
pbLiteral.setTimestampMillisValue(((TimestampNtz) literal).getMillisecond());

‎fluss-rpc/src/test/java/org/apache/fluss/rpc/util/PredicateMessageUtilsTest.java‎

Lines changed: 17 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,6 @@
5656

5757
import java.lang.reflect.Field;
5858
import java.math.BigDecimal;
59-
import java.time.LocalDate;
6059
import java.time.LocalTime;
6160
import java.util.Arrays;
6261
import java.util.Collections;
@@ -94,6 +93,20 @@ private static RowType buildRowType(LeafPredicate... predicates) {
9493
return buildRowType(Arrays.asList(predicates));
9594
}
9695

96+
@Test
97+
public void testLeafPredicateTimeIntegerLiteral() {
98+
int timeMillis = 45_000_000;
99+
DataType type = new TimeType(false, 3);
100+
LeafPredicate predicate =
101+
new LeafPredicate(
102+
Equal.INSTANCE, type, 0, "f_time", Collections.singletonList(timeMillis));
103+
RowType rowType = buildRowType(predicate);
104+
PbPredicate pb = PredicateMessageUtils.toPbPredicate(predicate, rowType);
105+
Predicate result = PredicateMessageUtils.toPredicate(pb, rowType);
106+
LeafPredicate lp = (LeafPredicate) result;
107+
assertThat(lp.literals().get(0)).isEqualTo(timeMillis);
108+
}
109+
97110
@Test
98111
public void testLeafPredicateIntEqual() {
99112
DataType type = new IntType(false);
@@ -317,22 +330,17 @@ public void testPbLiteralSerde() {
317330
DataType dateType = new DateType(false);
318331
LeafPredicate datePredicate =
319332
new LeafPredicate(
320-
Equal.INSTANCE,
321-
dateType,
322-
8,
323-
"f_date",
324-
Collections.singletonList(
325-
LocalDate.ofEpochDay(19000L))); // days since epoch
333+
Equal.INSTANCE, dateType, 8, "f_date", Collections.singletonList(19000));
326334
// time
327335
DataType timeType = new TimeType(false, 3);
328-
LocalTime time = LocalTime.of(12, 30);
336+
int timeMillis = (int) (LocalTime.of(12, 30).toNanoOfDay() / 1_000_000L);
329337
LeafPredicate timePredicate =
330338
new LeafPredicate(
331339
Equal.INSTANCE,
332340
timeType,
333341
9,
334342
"f_time",
335-
Collections.singletonList(time)); // millis of day
343+
Collections.singletonList(timeMillis));
336344
// timestamp
337345
DataType tsType = new TimestampType(false, 3);
338346
TimestampNtz ts = TimestampNtz.fromMillis(1680000000000L, 3);

‎fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/row/FlussAsSparkRow.scala‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
package org.apache.fluss.spark.row
1919

2020
import org.apache.fluss.row.{InternalRow => FlussInternalRow}
21-
import org.apache.fluss.types.{ArrayType => FlussArrayType, BinaryType => FlussBinaryType, LocalZonedTimestampType, MapType => FlussMapType, RowType, TimestampType}
21+
import org.apache.fluss.types.{ArrayType => FlussArrayType, BinaryType => FlussBinaryType, BytesType => FlussBytesType, LocalZonedTimestampType, MapType => FlussMapType, RowType, TimestampType}
2222
import org.apache.fluss.utils.InternalRowUtils
2323

2424
import org.apache.spark.sql.catalyst.{InternalRow => SparkInteralRow}
@@ -84,9 +84,9 @@ class FlussAsSparkRow(rowType: RowType) extends SparkInteralRow {
8484
DataConverter.toSparkUTF8String(row.getString(ordinal))
8585
}
8686

87-
override def getBinary(ordinal: Int): Array[Byte] = {
88-
val binaryType = rowType.getTypeAt(ordinal).asInstanceOf[FlussBinaryType]
89-
row.getBinary(ordinal, binaryType.getLength)
87+
override def getBinary(ordinal: Int): Array[Byte] = rowType.getTypeAt(ordinal) match {
88+
case b: FlussBinaryType => row.getBinary(ordinal, b.getLength)
89+
case _: FlussBytesType => row.getBytes(ordinal)
9090
}
9191

9292
override def getInterval(ordinal: Int): CalendarInterval =

‎fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/utils/SparkPredicateConverter.scala‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ import org.apache.spark.sql.types.{Decimal => SparkDecimal}
2727
import org.apache.spark.unsafe.types.UTF8String
2828

2929
import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float => JFloat, Long => JLong, Short => JShort}
30-
import java.time.LocalDate
3130

3231
import scala.jdk.CollectionConverters._
3332

@@ -277,9 +276,8 @@ object SparkPredicateConverter {
277276
}
278277

279278
case DataTypeRoot.DATE =>
280-
// RPC serialization (PredicateMessageUtils) expects LocalDate.
281279
value match {
282-
case d: Integer => LocalDate.ofEpochDay(d.longValue())
280+
case d: Integer => d
283281
case _ => throw new UnsupportedExpression
284282
}
285283

‎fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala‎

Lines changed: 29 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -498,15 +498,38 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
498498
}
499499
}
500500

501-
test("Spark Read: partition pushdown — INT partition with range predicate") {
501+
test("Spark Read: partition pushdown — all supported types") {
502502
withTable("t") {
503503
sql(s"""
504-
|CREATE TABLE $DEFAULT_DATABASE.t (id BIGINT, pt INT)
505-
|PARTITIONED BY (pt)""".stripMargin)
506-
sql(s"INSERT INTO $DEFAULT_DATABASE.t VALUES (1, 2), (2, 10)")
504+
|CREATE TABLE $DEFAULT_DATABASE.t (
505+
| id INT,
506+
| p_bool BOOLEAN, p_int_eq INT, p_int_range INT,
507+
| p_bigint BIGINT, p_float FLOAT, p_double DOUBLE,
508+
| p_string STRING, p_binary BINARY,
509+
| p_date DATE, p_ts TIMESTAMP, p_ts_ntz TIMESTAMP_NTZ)
510+
|PARTITIONED BY (
511+
| p_bool, p_int_eq, p_int_range, p_bigint, p_float, p_double,
512+
| p_string, p_binary, p_date, p_ts, p_ts_ntz)""".stripMargin)
513+
sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES
514+
|(1, false, 10, 10, 99999L, CAST(12.5 AS FLOAT), 7.88, 'hello',
515+
| CAST('Hi' AS BINARY), DATE '2026-01-01',
516+
| TIMESTAMP '2026-01-01 12:00:00', TIMESTAMP_NTZ '2026-01-01 12:00:00'),
517+
|(2, true, 11, 2, 99998L, CAST(13.5 AS FLOAT), 8.88, 'world',
518+
| CAST('Bye' AS BINARY), DATE '2026-01-02',
519+
| TIMESTAMP '2026-01-02 12:00:00', TIMESTAMP_NTZ '2026-01-02 12:00:00')
520+
|""".stripMargin)
507521

508-
val query = sql(s"SELECT * FROM $DEFAULT_DATABASE.t WHERE pt > 2 ORDER BY id")
509-
checkAnswer(query, Row(2L, 10) :: Nil)
522+
val query = sql(s"""
523+
|SELECT id FROM $DEFAULT_DATABASE.t WHERE
524+
| p_bool = false AND p_int_eq = 10 AND p_int_range > 2
525+
| AND p_bigint = 99999L
526+
| AND p_float = CAST(12.5 AS FLOAT) AND p_double = 7.88
527+
| AND p_string = 'hello' AND p_binary = CAST('Hi' AS BINARY)
528+
| AND p_date = DATE '2026-01-01'
529+
| AND p_ts = TIMESTAMP '2026-01-01 12:00:00'
530+
| AND p_ts_ntz = TIMESTAMP_NTZ '2026-01-01 12:00:00'
531+
|""".stripMargin)
532+
checkAnswer(query, Row(1) :: Nil)
510533
assert(partitionPredicate(query).isDefined)
511534
}
512535
}

‎fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/utils/SparkPredicateConverterTest.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -195,7 +195,7 @@ class SparkPredicateConverterTest extends AnyFunSuite {
195195
test("Date literal from epoch days") {
196196
val days = Integer.valueOf(LocalDate.of(2025, 1, 15).toEpochDay.toInt)
197197
val predicate = convert(pred("=", ref("dt"), lit(days, DateType)))
198-
val expected = new PredicateBuilder(rowType).equal(6, LocalDate.of(2025, 1, 15))
198+
val expected = new PredicateBuilder(rowType).equal(6, days)
199199
assertThat(predicate).isEqualTo(expected)
200200
}
201201

0 commit comments

Comments
 (0)