This is an automated email from the ASF dual-hosted git repository.
czweng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b99f8e663 [flink] Lookup source should ignore scan options (#4383)
b99f8e663 is described below
commit b99f8e663a211d655e459152cb399844e2577d38
Author: yuzelin <[email protected]>
AuthorDate: Mon Oct 28 19:05:26 2024 +0800
[flink] Lookup source should ignore scan options (#4383)
This closes #4383.
---
.../paimon/flink/source/BaseDataTableSource.java | 51 +++++++++++++--
.../org/apache/paimon/flink/LookupJoinITCase.java | 73 ++++++++++------------
2 files changed, 80 insertions(+), 44 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
index 9458f7817..5dbbdcedd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
@@ -29,9 +29,12 @@ import org.apache.paimon.flink.log.LogStoreTableFactory;
import org.apache.paimon.flink.lookup.FileStoreLookupFunction;
import org.apache.paimon.flink.lookup.LookupRuntimeProviderFactory;
import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.utils.Projection;
@@ -41,10 +44,6 @@ import org.apache.flink.api.connector.source.Source;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.source.LookupTableSource;
-import org.apache.flink.table.connector.source.LookupTableSource.LookupContext;
-import
org.apache.flink.table.connector.source.LookupTableSource.LookupRuntimeProvider;
-import org.apache.flink.table.connector.source.ScanTableSource.ScanContext;
-import
org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
import org.apache.flink.table.connector.source.SourceProvider;
import
org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
@@ -56,7 +55,10 @@ import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.IntStream;
@@ -80,6 +82,16 @@ import static
org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_IDLE_
public abstract class BaseDataTableSource extends FlinkTableSource
implements LookupTableSource, SupportsWatermarkPushDown,
SupportsAggregatePushDown {
+ private static final List<ConfigOption<?>> TIME_TRAVEL_OPTIONS =
+ Arrays.asList(
+ CoreOptions.SCAN_TIMESTAMP,
+ CoreOptions.SCAN_TIMESTAMP_MILLIS,
+ CoreOptions.SCAN_WATERMARK,
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_SNAPSHOT_ID,
+ CoreOptions.SCAN_TAG_NAME,
+ CoreOptions.SCAN_VERSION);
+
protected final ObjectIdentifier tableIdentifier;
protected final boolean streaming;
protected final DynamicTableFactory.Context context;
@@ -231,6 +243,12 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
@Override
public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext
context) {
+ if (!(table instanceof FileStoreTable)) {
+ throw new UnsupportedOperationException(
+ "Currently, lookup dim table only support FileStoreTable
but is "
+ + table.getClass().getName());
+ }
+
if (limit != null) {
throw new RuntimeException(
"Limit push down should not happen in Lookup source, but
it is " + limit);
@@ -244,11 +262,34 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
boolean enableAsync = options.get(LOOKUP_ASYNC);
int asyncThreadNumber = options.get(LOOKUP_ASYNC_THREAD_NUMBER);
return LookupRuntimeProviderFactory.create(
- new FileStoreLookupFunction(table, projection, joinKey,
predicate),
+ getFileStoreLookupFunction(
+ context,
+ timeTravelDisabledTable((FileStoreTable) table),
+ projection,
+ joinKey),
enableAsync,
asyncThreadNumber);
}
+ protected FileStoreLookupFunction getFileStoreLookupFunction(
+ LookupContext context, Table table, int[] projection, int[]
joinKey) {
+ return new FileStoreLookupFunction(table, projection, joinKey,
predicate);
+ }
+
+ private FileStoreTable timeTravelDisabledTable(FileStoreTable table) {
+ Map<String, String> newOptions = new HashMap<>(table.options());
+
TIME_TRAVEL_OPTIONS.stream().map(ConfigOption::key).forEach(newOptions::remove);
+
+ CoreOptions.StartupMode startupMode =
CoreOptions.fromMap(newOptions).startupMode();
+ if (startupMode != CoreOptions.StartupMode.COMPACTED_FULL) {
+ startupMode = CoreOptions.StartupMode.LATEST_FULL;
+ }
+ newOptions.put(CoreOptions.SCAN_MODE.key(), startupMode.toString());
+
+ TableSchema newSchema = table.schema().copy(newOptions);
+ return table.copy(newSchema);
+ }
+
@Override
public boolean applyAggregates(
List<int[]> groupingSets,
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
index 46399f856..3e9ba2194 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
@@ -134,45 +134,40 @@ public class LookupJoinITCase extends CatalogITCaseBase {
iterator.close();
}
- @ParameterizedTest
- @EnumSource(LookupCacheMode.class)
- public void testLookupIgnoreScanOptions(LookupCacheMode cacheMode) throws
Exception {
- initTable(cacheMode);
- sql("INSERT INTO DIM VALUES (1, 11, 111, 1111), (2, 22, 222, 2222)");
-
- String scanOption;
- if (ThreadLocalRandom.current().nextBoolean()) {
- scanOption = "'scan.mode'='latest'";
- } else {
- scanOption = "'scan.snapshot-id'='2'";
- }
- String query =
- "SELECT T.i, D.j, D.k1, D.k2 FROM T LEFT JOIN DIM /*+ OPTIONS("
- + scanOption
- + ") */"
- + " for system_time as of T.proctime AS D ON T.i =
D.i";
- BlockingIterator<Row, Row> iterator =
BlockingIterator.of(sEnv.executeSql(query).collect());
-
- sql("INSERT INTO T VALUES (1), (2), (3)");
- List<Row> result = iterator.collect(3);
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111, 1111),
- Row.of(2, 22, 222, 2222),
- Row.of(3, null, null, null));
-
- sql("INSERT INTO DIM VALUES (2, 44, 444, 4444), (3, 33, 333, 3333)");
- Thread.sleep(2000); // wait refresh
- sql("INSERT INTO T VALUES (1), (2), (3), (4)");
- result = iterator.collect(4);
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111, 1111),
- Row.of(2, 44, 444, 4444),
- Row.of(3, 33, 333, 3333),
- Row.of(4, null, null, null));
-
- iterator.close();
+ @Test
+ public void testLookupIgnoreScanOptions() throws Exception {
+ sql(
+ "CREATE TABLE d (\n"
+ + " pt INT,\n"
+ + " id INT,\n"
+ + " data STRING,\n"
+ + " PRIMARY KEY (pt, id) NOT ENFORCED\n"
+ + ") PARTITIONED BY (pt) WITH ( 'bucket' = '1',
'continuous.discovery-interval'='1 ms' )");
+ sql(
+ "CREATE TABLE t1 (\n"
+ + " pt INT,\n"
+ + " id INT,\n"
+ + " data STRING,\n"
+ + " `proctime` AS PROCTIME(),\n"
+ + " PRIMARY KEY (pt, id) NOT ENFORCED\n"
+ + ") PARTITIONED BY (pt) with (
'continuous.discovery-interval'='1 ms' )");
+
+ sql("INSERT INTO d VALUES (1, 1, 'one'), (2, 2, 'two'), (3, 3,
'three')");
+ sql("INSERT INTO t1 VALUES (1, 1, 'one'), (2, 2, 'two'), (3, 3,
'three')");
+
+ BlockingIterator<Row, Row> streamIter =
+ streamSqlBlockIter(
+ "SELECT T.pt, T.id, T.data, D.pt, D.id, D.data "
+ + "FROM t1 AS T LEFT JOIN d /*+
OPTIONS('lookup.dynamic-partition'='max_pt()', 'scan.snapshot-id'='2') */ "
+ + "FOR SYSTEM_TIME AS OF T.proctime AS D ON
T.id = D.id");
+
+ assertThat(streamIter.collect(3))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, "one", null, null, null),
+ Row.of(2, 2, "two", null, null, null),
+ Row.of(3, 3, "three", 3, 3, "three"));
+
+ streamIter.close();
}
@ParameterizedTest
