This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new e9a7f39661 Flink: Support Lookup Join using full in-memory lookup
cache (#18144)
e9a7f39661 is described below
commit e9a7f39661a6c0cf77338db3aebdc7be59790859
Author: GuoYu <[email protected]>
AuthorDate: Tue Nov 3 16:51:47 2026 +0800
Flink: Support Lookup Join using full in-memory lookup cache (#18144)
---
docs/docs/flink-configuration.md | 8 +
docs/docs/flink-queries.md | 22 ++
.../iceberg/flink/FlinkDynamicTableFactory.java | 2 +
.../iceberg/flink/source/IcebergTableSource.java | 60 +++-
.../lookup/IcebergFullCachingLookupFunction.java | 223 +++++++++++++
.../flink/source/lookup/IcebergLookupOptions.java | 41 +++
.../flink/source/lookup/IcebergLookupReader.java | 124 +++++++
.../flink/source/lookup/InMemoryLookupCache.java | 69 ++++
.../iceberg/flink/source/TestFlinkSourceSql.java | 2 +-
.../flink/source/TestIcebergLookupJoinSql.java | 369 +++++++++++++++++++++
.../iceberg/flink/source/TestIcebergSourceSql.java | 2 +-
.../{TestSqlBase.java => TestSourceSqlBase.java} | 54 +--
.../apache/iceberg/flink/source/TestSqlBase.java | 97 +-----
.../flink/source/lookup/TestIcebergLookup.java | 318 ++++++++++++++++++
14 files changed, 1239 insertions(+), 152 deletions(-)
diff --git a/docs/docs/flink-configuration.md b/docs/docs/flink-configuration.md
index d0b4889177..a3d8626770 100644
--- a/docs/docs/flink-configuration.md
+++ b/docs/docs/flink-configuration.md
@@ -226,3 +226,11 @@ builder.
| table.exec.iceberg.fetch-batch-record-count | 2048
| Target number of records per fetch batch in the Iceberg source
reader.
|
| table.exec.iceberg.worker-pool-size | max(2, available cpu)
| Size of the worker pool used to plan or scan manifests. Defaults to the
shared Iceberg worker pool size, which is controlled by the
`iceberg.worker.num-threads` system property.
|
| table.exec.iceberg.use-v2-sink | false
| Use the SinkV2 based `IcebergSink` implementation, see [Sink V2
based implementation](flink-writes.md#sink-v2-based-implementation).
|
+
+### Lookup options
+
+Flink lookup joins cache the whole projected dimension table in memory, and
the cache is kept in memory for the lifetime of the job, so the dimension table
should be populated before the join starts. See [Lookup
Join](flink-queries.md#lookup-join) for details. These options are set as table
options in the DDL, or per query with the `OPTIONS` hint.
+
+| Lookup option | Default | Description
|
+| ------------------------------ | ------- |
--------------------------------------------------------------------------------------------------
|
+| `lookup.full-cache.eager-load` | `true` | Whether to load the full cache
when the lookup function is opened, instead of on the first lookup. |
diff --git a/docs/docs/flink-queries.md b/docs/docs/flink-queries.md
index 8518f2f403..15723ef863 100644
--- a/docs/docs/flink-queries.md
+++ b/docs/docs/flink-queries.md
@@ -92,6 +92,28 @@ SELECT * FROM table /*+ OPTIONS('tag'='t1') */;
SELECT * FROM table /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s',
'start-tag'='t1', 'end-tag'='t2') */;
```
+### Lookup Join
+
+Iceberg supports Flink lookup join, which enriches a stream with data from an
Iceberg dimension table:
+
+```sql
+SELECT o.order_id, o.user_id, u.name, u.city
+FROM orders AS o
+LEFT JOIN iceberg_catalog.db.user_dim
+ FOR SYSTEM_TIME AS OF o.proc_time AS u
+ ON o.user_id = u.user_id;
+```
+
+Iceberg implements lookup join with a full cache: the whole projected
dimension table is loaded into the cache, and every lookup is served from it
without falling back to the table. The full cache is held in memory on the
TaskManager heap, so lookup join targets dimension tables that fit comfortably
there.
+
+`lookup.cache` can be left unset. If it is set, the only supported value is
`FULL`, and it must be upper case: unlike in Flink's built-in connectors,
`'lookup.cache'='full'` is rejected.
+
+`lookup.full-cache.eager-load` decides whether the job blocks at startup or on
the first lookup. With the default `true`, the lookup function loads the cache
when it is opened, so the job blocks during deployment, before it processes any
data, and fails at startup if the dimension table cannot be read; with `false`,
the load is deferred to the first lookup, so the data flow blocks only when the
first probe row arrives, and the subtasks of the join can end up on different
snapshots of the [...]
+
+There is no background refresh: each subtask keeps the snapshot it loaded, so
subtasks can serve different snapshots of the dimension table. Populate the
dimension table before the join starts.
+
+Check out all the options here:
[lookup-options](flink-configuration.md#lookup-options)
+
## Reading with DataStream
Iceberg support streaming or batch read in Java API now.
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index 6306ee7a0a..0c7244dc23 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
@@ -39,6 +39,7 @@ import
org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.flink.source.IcebergTableSource;
+import org.apache.iceberg.flink.source.lookup.IcebergLookupOptions;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
@@ -132,6 +133,7 @@ public class FlinkDynamicTableFactory
options.add(FlinkCreateTableOptions.CATALOG_TABLE);
options.add(FlinkCreateTableOptions.USE_DYNAMIC_ICEBERG_SINK);
options.add(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+ options.add(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD);
return options;
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
index 40df25b59f..759b316667 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
@@ -33,22 +33,30 @@ import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.ProviderContext;
import org.apache.flink.table.connector.source.DataStreamScanProvider;
import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsSourceWatermark;
+import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupOptions;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.LookupFunction;
import org.apache.flink.table.legacy.api.TableSchema;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkConfParser;
import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.FlinkFilters;
import org.apache.iceberg.flink.FlinkReadOptions;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.source.assigner.SplitAssignerType;
+import org.apache.iceberg.flink.source.lookup.IcebergFullCachingLookupFunction;
+import org.apache.iceberg.flink.source.lookup.IcebergLookupOptions;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -61,7 +69,8 @@ public class IcebergTableSource
SupportsProjectionPushDown,
SupportsFilterPushDown,
SupportsLimitPushDown,
- SupportsSourceWatermark {
+ SupportsSourceWatermark,
+ LookupTableSource {
private int[] projectedFields;
private Long limit;
@@ -72,6 +81,7 @@ public class IcebergTableSource
private final Map<String, String> properties;
private final boolean isLimitPushDown;
private final ReadableConfig readableConfig;
+ private final FlinkConfParser flinkConfParser;
private IcebergTableSource(IcebergTableSource toCopy) {
this.loader = toCopy.loader;
@@ -82,6 +92,7 @@ public class IcebergTableSource
this.limit = toCopy.limit;
this.filters = toCopy.filters;
this.readableConfig = toCopy.readableConfig;
+ this.flinkConfParser = toCopy.flinkConfParser;
}
public IcebergTableSource(
@@ -109,6 +120,7 @@ public class IcebergTableSource
this.limit = limit;
this.filters = filters;
this.readableConfig = readableConfig;
+ this.flinkConfParser = new FlinkConfParser(properties, readableConfig);
}
@Override
@@ -237,4 +249,50 @@ public class IcebergTableSource
public String asSummaryString() {
return "Iceberg table source";
}
+
+ @Override
+ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context)
{
+ int[][] lookupKeys = context.getKeys();
+ int[] lookupKeyIndexes = new int[lookupKeys.length];
+ for (int i = 0; i < lookupKeys.length; i++) {
+ Preconditions.checkArgument(
+ lookupKeys[i].length == 1, "Iceberg lookup source doesn't support
nested lookup key.");
+ lookupKeyIndexes[i] = lookupKeys[i][0];
+ }
+
+ ResolvedSchema projected = getProjectedSchema();
+ RowType projectedRowType = (RowType)
projected.toPhysicalRowDataType().getLogicalType();
+
+ LookupOptions.LookupCacheType requestedCacheType =
+ flinkConfParser
+ .enumConfParser(LookupOptions.LookupCacheType.class)
+ .option(LookupOptions.CACHE_TYPE.key())
+ .parseOptional();
+ Preconditions.checkArgument(
+ requestedCacheType == null || requestedCacheType ==
LookupOptions.LookupCacheType.FULL,
+ "Iceberg lookup join only supports %s=FULL, but it is set to '%s'.
NONE and PARTIAL are "
+ + "not supported, because an Iceberg table cannot be
point-looked-up.",
+ LookupOptions.CACHE_TYPE.key(),
+ requestedCacheType);
+
+ boolean eagerLoad =
+ flinkConfParser
+ .booleanConf()
+ .option(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD.key())
+
.defaultValue(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD.defaultValue())
+ .parse();
+
+ boolean caseSensitive =
+ flinkConfParser
+ .booleanConf()
+ .option(FlinkReadOptions.CASE_SENSITIVE)
+ .flinkConfig(FlinkReadOptions.CASE_SENSITIVE_OPTION)
+
.defaultValue(FlinkReadOptions.CASE_SENSITIVE_OPTION.defaultValue())
+ .parse();
+
+ LookupFunction lookupFn =
+ new IcebergFullCachingLookupFunction(
+ loader, projectedRowType, lookupKeyIndexes, filters,
caseSensitive, eagerLoad);
+ return LookupFunctionProvider.of(lookupFn);
+ }
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
new file mode 100644
index 0000000000..d7e78913e3
--- /dev/null
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.lookup;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.LookupFunction;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkRowData;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.io.CloseableIterable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A lookup function that serves every lookup from an in-memory cache of the
rows of an Iceberg
+ * table, after projection and pushed-down filters are applied.
+ *
+ * <p>The cache is loaded from the current snapshot either when the function
is opened or on the
+ * first lookup, depending on {@link
IcebergLookupOptions#FULL_CACHE_EAGER_LOAD}. It is never
+ * refreshed, so later changes to the table are not visible.
+ */
+@Internal
+public class IcebergFullCachingLookupFunction extends LookupFunction {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergFullCachingLookupFunction.class);
+
+ private static final String METRIC_GROUP = "cache";
+ private static final long UNINITIALIZED = -1L;
+
+ private final TableLoader tableLoader;
+ private final RowType projectedRowType;
+ private final int[] lookupKeyIndexes;
+ private final List<Expression> pushedFilters;
+ private final boolean caseSensitive;
+ private final boolean eagerLoad;
+
+ private transient Table table;
+ private transient IcebergLookupReader reader;
+ private transient RowDataSerializer rowSerializer;
+ private transient RowDataSerializer keySerializer;
+ private transient RowData.FieldGetter[] keyGetters;
+ private transient InMemoryLookupCache cache;
+
+ private transient Counter missCounter;
+ private transient volatile long currentSnapshotId;
+ private transient volatile long cachedRecords;
+ private transient volatile long latestLoadTimeMs;
+
+ public IcebergFullCachingLookupFunction(
+ TableLoader tableLoader,
+ RowType projectedRowType,
+ int[] lookupKeyIndexes,
+ List<Expression> pushedFilters,
+ boolean caseSensitive,
+ boolean eagerLoad) {
+ this.tableLoader = tableLoader;
+ this.projectedRowType = projectedRowType;
+ this.lookupKeyIndexes = lookupKeyIndexes;
+ this.pushedFilters = pushedFilters;
+ this.caseSensitive = caseSensitive;
+ this.eagerLoad = eagerLoad;
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ super.open(context);
+ LOG.info(
+ "IcebergFullCachingLookupFunction opening, projected fields={},
lookupKeyIndexes={}",
+ projectedRowType.getFieldNames(),
+ Arrays.toString(lookupKeyIndexes));
+
+ MetricGroup metricGroup = context.getMetricGroup().addGroup(METRIC_GROUP);
+ this.missCounter = metricGroup.counter("missCount");
+ this.currentSnapshotId = IcebergLookupReader.CURRENT_SNAPSHOT;
+ this.cachedRecords = 0L;
+ this.latestLoadTimeMs = UNINITIALIZED;
+ metricGroup.gauge("snapshotId", () -> currentSnapshotId);
+ metricGroup.gauge("numCachedRecords", () -> cachedRecords);
+ metricGroup.gauge("latestLoadTime", () -> latestLoadTimeMs);
+
+ tableLoader.open();
+ this.table = tableLoader.loadTable();
+
+ Schema icebergProjection =
+ FlinkSchemaUtil.convert(table.schema(),
FlinkSchemaUtil.toResolvedSchema(projectedRowType));
+
+ createAccessors();
+
+ String nameMapping =
table.properties().get(TableProperties.DEFAULT_NAME_MAPPING);
+ this.reader =
+ new IcebergLookupReader(
+ table, icebergProjection, pushedFilters, caseSensitive,
nameMapping);
+
+ if (eagerLoad) {
+ loadCache();
+ }
+ }
+
+ /**
+ * Returns the cached rows for the key. The returned collection and its rows
are shared with the
+ * cache, so they must not be modified.
+ */
+ @Override
+ public Collection<RowData> lookup(RowData keyRow) throws IOException {
+ if (cache == null) {
+ loadCache();
+ }
+
+ List<RowData> hit = cache.get(keySerializer.toBinaryRow(keyRow));
+ if (hit == null) {
+ missCounter.inc();
+ return Collections.emptyList();
+ }
+
+ return hit;
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ tableLoader.close();
+ } finally {
+ if (cache != null) {
+ cache.close();
+ }
+
+ super.close();
+ }
+ }
+
+ private void loadCache() throws IOException {
+ table.refresh();
+ Snapshot snapshot = table.currentSnapshot();
+ long snapshotId =
+ snapshot == null ? IcebergLookupReader.CURRENT_SNAPSHOT :
snapshot.snapshotId();
+
+ LOG.info(
+ "IcebergFullCachingLookupFunction loading started, snapshot={},
committedAt={}, projected fields={}, pushedFilters={}",
+ snapshot == null ? "none" : snapshotId,
+ snapshot == null ? "n/a" : snapshot.timestampMillis(),
+ projectedRowType.getFieldNames(),
+ pushedFilters);
+
+ InMemoryLookupCache loaded = new InMemoryLookupCache();
+ long rowCnt = 0L;
+ long start = System.currentTimeMillis();
+ try (CloseableIterable<RowData> rows = reader.read(snapshotId)) {
+ for (RowData row : rows) {
+ loaded.add(extractKey(row), rowSerializer.toBinaryRow(row).copy());
+ rowCnt++;
+ }
+ } finally {
+ this.latestLoadTimeMs = System.currentTimeMillis() - start;
+ }
+
+ this.cache = loaded;
+ this.currentSnapshotId = snapshotId;
+ this.cachedRecords = rowCnt;
+
+ LOG.info(
+ "IcebergFullCachingLookupFunction loading finished, snapshot={},
rows={}, cost={} ms",
+ snapshot == null ? "none" : snapshotId,
+ rowCnt,
+ latestLoadTimeMs);
+ }
+
+ private BinaryRowData extractKey(RowData row) {
+ GenericRowData key = new GenericRowData(keyGetters.length);
+ for (int i = 0; i < keyGetters.length; i++) {
+ key.setField(i, keyGetters[i].getFieldOrNull(row));
+ }
+
+ return keySerializer.toBinaryRow(key).copy();
+ }
+
+ private void createAccessors() {
+ LogicalType[] keyTypes = new LogicalType[lookupKeyIndexes.length];
+ this.keyGetters = new RowData.FieldGetter[lookupKeyIndexes.length];
+ for (int i = 0; i < lookupKeyIndexes.length; i++) {
+ int projectedIndex = lookupKeyIndexes[i];
+ keyTypes[i] = projectedRowType.getTypeAt(projectedIndex);
+ this.keyGetters[i] = FlinkRowData.createFieldGetter(keyTypes[i],
projectedIndex);
+ }
+
+ this.rowSerializer = new RowDataSerializer(projectedRowType);
+ this.keySerializer = new RowDataSerializer(keyTypes);
+ }
+}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
new file mode 100644
index 0000000000..355fd5387f
--- /dev/null
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.lookup;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/** Lookup join options for the Iceberg table source. */
+@Internal
+public class IcebergLookupOptions {
+
+ public static final ConfigOption<Boolean> FULL_CACHE_EAGER_LOAD =
+ ConfigOptions.key("lookup.full-cache.eager-load")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to load the full lookup cache when the lookup function
is opened, instead "
+ + "of on the first lookup. Eager loading narrows the window
in which the subtasks "
+ + "of the join can end up on different snapshots of the
dimension table, and "
+ + "fails the job at startup if the table cannot be read, at
the cost of a longer "
+ + "deployment.");
+
+ private IcebergLookupOptions() {}
+}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
new file mode 100644
index 0000000000..d669178557
--- /dev/null
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.lookup;
+
+import java.io.IOException;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.flink.source.DataIterator;
+import org.apache.iceberg.flink.source.FileScanTaskReader;
+import org.apache.iceberg.flink.source.RowDataFileScanTaskReader;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.util.SnapshotUtil;
+
+/** Reads all rows of an Iceberg table to build a lookup cache. */
+@Internal
+class IcebergLookupReader {
+
+ static final long CURRENT_SNAPSHOT = -1L;
+
+ private final Table table;
+ private final Schema projectedSchema;
+ private final List<Expression> baseFilters;
+ private final Expression filter;
+ private final boolean caseSensitive;
+ private final String nameMapping;
+
+ IcebergLookupReader(
+ Table table,
+ Schema projectedSchema,
+ List<Expression> baseFilters,
+ boolean caseSensitive,
+ @Nullable String nameMapping) {
+ Preconditions.checkNotNull(baseFilters, "Base filters should not be null");
+
+ this.table = table;
+ this.projectedSchema = projectedSchema;
+ this.baseFilters = baseFilters;
+ this.caseSensitive = caseSensitive;
+ this.nameMapping = nameMapping;
+
+ Expression combinedFilter = Expressions.alwaysTrue();
+ for (Expression baseFilter : this.baseFilters) {
+ combinedFilter = Expressions.and(combinedFilter, baseFilter);
+ }
+
+ this.filter = combinedFilter;
+ }
+
+ CloseableIterable<RowData> read(long snapshotId) {
+ Schema tableSchema =
+ snapshotId == CURRENT_SNAPSHOT ? table.schema() :
SnapshotUtil.schemaFor(table, snapshotId);
+
+ RowDataFileScanTaskReader fileReader =
+ new RowDataFileScanTaskReader(
+ tableSchema, projectedSchema, nameMapping, caseSensitive,
baseFilters);
+
+ TableScan scan =
+
table.newScan().caseSensitive(caseSensitive).project(projectedSchema).filter(filter);
+ if (snapshotId != CURRENT_SNAPSHOT) {
+ scan = scan.useSnapshot(snapshotId);
+ }
+
+ return new LookupIterable(scan.planTasks(), fileReader);
+ }
+
+ private class LookupIterable extends CloseableGroup implements
CloseableIterable<RowData> {
+ private final CloseableIterable<CombinedScanTask> tasks;
+ private final FileScanTaskReader<RowData> fileReader;
+
+ private LookupIterable(
+ CloseableIterable<CombinedScanTask> tasks, FileScanTaskReader<RowData>
fileReader) {
+ this.tasks = tasks;
+ this.fileReader = fileReader;
+ }
+
+ @Override
+ public CloseableIterator<RowData> iterator() {
+ CloseableIterator<RowData> rows =
+ CloseableIterable.concat(Iterables.transform(tasks,
this::taskRows)).iterator();
+ addCloseable(rows);
+ return rows;
+ }
+
+ @Override
+ public void close() throws IOException {
+ tasks.close();
+ super.close();
+ }
+
+ private CloseableIterable<RowData> taskRows(CombinedScanTask task) {
+ DataIterator<RowData> rows =
+ new DataIterator<>(fileReader, task, table.io(), table.encryption());
+ return CloseableIterable.combine(() -> rows, rows);
+ }
+ }
+}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
new file mode 100644
index 0000000000..b6e1a4ebc2
--- /dev/null
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.lookup;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+class InMemoryLookupCache implements Closeable {
+
+ private final Map<BinaryRowData, List<RowData>> cache;
+
+ InMemoryLookupCache() {
+ this.cache = Maps.newHashMap();
+ }
+
+ @Nullable
+ List<RowData> get(BinaryRowData key) {
+ return cache.get(key);
+ }
+
+ void add(BinaryRowData key, BinaryRowData row) {
+ cache.compute(key, (k, existing) -> append(existing, row));
+ }
+
+ private static List<RowData> append(@Nullable List<RowData> existing,
RowData row) {
+ if (existing == null) {
+ return Collections.singletonList(row);
+ }
+
+ if (existing instanceof ArrayList) {
+ existing.add(row);
+ return existing;
+ }
+
+ List<RowData> rows = Lists.newArrayListWithCapacity(2);
+ rows.add(existing.get(0));
+ rows.add(row);
+ return rows;
+ }
+
+ @Override
+ public void close() {
+ cache.clear();
+ }
+}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
index 2dc5bc5c65..4a3d585e9d 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
@@ -37,7 +37,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
/** Use the FlinkSource */
-public class TestFlinkSourceSql extends TestSqlBase {
+public class TestFlinkSourceSql extends TestSourceSqlBase {
@BeforeEach
@Override
public void before() throws IOException {
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
new file mode 100644
index 0000000000..40c13af6d4
--- /dev/null
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
@@ -0,0 +1,369 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Tests SQL lookup join with Iceberg table source. */
+class TestIcebergLookupJoinSql extends TestSqlBase {
+ private static final Schema DIM_SCHEMA =
+ new Schema(
+ required(1, "user_id", Types.LongType.get()),
+ required(2, "name", Types.StringType.get()),
+ required(3, "city", Types.StringType.get()));
+
+ @BeforeEach
+ @Override
+ public void before() throws IOException {
+ setUpTableEnv(getStreamingTableEnv());
+ }
+
+ @Test
+ void lookupJoinReturnsMatchedAndUnmatchedRows() throws Exception {
+ createDefaultDimTable();
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders", 5);
+
+ String joinSql = lookupJoinSql("o.order_id, o.user_id, u.name, u.city",
"orders");
+
+ assertThat(SqlHelpers.sql(streamEnv, joinSql))
+ .containsExactlyInAnyOrder(
+ Row.of(1L, 1L, "alice", "beijing"),
+ Row.of(2L, 2L, "bob", "shanghai"),
+ Row.of(3L, 3L, "carol", "guangzhou"),
+ Row.of(4L, 4L, null, null),
+ Row.of(5L, 5L, null, null));
+ }
+
+ @Test
+ void lookupJoinWithFilterConditionInJoin() throws Exception {
+ Table dimTable =
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER,
DIM_SCHEMA);
+ GenericAppenderHelper helper =
+ new GenericAppenderHelper(dimTable, FileFormat.PARQUET,
temporaryFolder);
+ helper.appendToTable(
+ helper.writeFile(
+ Lists.newArrayList(
+ dimRecord(1L, "alice", "beijing"), dimRecord(3L, "carol",
"beijing"))));
+ DataFile shanghaiFile = helper.writeFile(Lists.newArrayList(dimRecord(2L,
"bob", "shanghai")));
+ helper.appendToTable(shanghaiFile);
+
+ // The filter must be pushed to the dimension table scan, which skips this
file by its
+ // column stats. Otherwise loading the cache fails on the missing file.
+ dimTable.io().deleteFile(shanghaiFile.location());
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders_with_filter", 4);
+
+ String joinSql =
+ lookupJoinSql(
+ "o.order_id, o.user_id, u.name, u.city",
+ "orders_with_filter",
+ null,
+ "u.city = 'beijing'");
+
+ assertThat(SqlHelpers.sql(streamEnv, joinSql))
+ .containsExactlyInAnyOrder(
+ Row.of(1L, 1L, "alice", "beijing"),
+ Row.of(2L, 2L, null, null),
+ Row.of(3L, 3L, "carol", "beijing"),
+ Row.of(4L, 4L, null, null));
+ }
+
+ @Test
+ void lookupJoinWithDuplicateKeysInDimTable() throws Exception {
+ // The dimension table has two rows sharing the same join key, so every
probe row
+ // matching that key must be joined with both of them.
+ createDimTable(
+ dimRecord(1L, "alice", "beijing"),
+ dimRecord(1L, "alice-2", "shanghai"),
+ dimRecord(2L, "bob", "shanghai"));
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders_duplicate_keys", 3);
+
+ String joinSql =
+ lookupJoinSql("o.order_id, o.user_id, u.name, u.city",
"orders_duplicate_keys");
+
+ assertThat(SqlHelpers.sql(streamEnv, joinSql))
+ .containsExactlyInAnyOrder(
+ Row.of(1L, 1L, "alice", "beijing"),
+ Row.of(1L, 1L, "alice-2", "shanghai"),
+ Row.of(2L, 2L, "bob", "shanghai"),
+ Row.of(3L, 3L, null, null));
+ }
+
+ @Test
+ void lookupJoinWithOptionsInTableDdl() throws Exception {
+ createDefaultDimTable();
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders_ddl", 3);
+
+ SqlHelpers.sql(
+ streamEnv,
+ "CREATE TABLE default_catalog.default_database.dim_users_ddl (\n"
+ + " user_id BIGINT,\n"
+ + " name STRING,\n"
+ + " city STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = 'iceberg',\n"
+ + " 'catalog-name' = 'iceberg_catalog',\n"
+ + " 'catalog-type' = 'hadoop',\n"
+ + " 'warehouse' = '%s',\n"
+ + " 'catalog-database' = '%s',\n"
+ + " 'catalog-table' = '%s',\n"
+ + " 'lookup.cache' = 'FULL'\n"
+ + ")",
+ CATALOG_EXTENSION.warehouse(),
+ TestFixtures.DATABASE,
+ TestFixtures.TABLE);
+
+ String joinSql =
+ "SELECT o.order_id, o.user_id, u.name, u.city\n"
+ + "FROM orders_ddl AS o\n"
+ + "LEFT JOIN default_catalog.default_database.dim_users_ddl\n"
+ + " FOR SYSTEM_TIME AS OF o.proc_time AS u\n"
+ + " ON o.user_id = u.user_id";
+
+ assertThat(SqlHelpers.sql(streamEnv, joinSql))
+ .containsExactlyInAnyOrder(
+ Row.of(1L, 1L, "alice", "beijing"),
+ Row.of(2L, 2L, "bob", "shanghai"),
+ Row.of(3L, 3L, "carol", "guangzhou"));
+ }
+
+ @Test
+ void lookupJoinAcceptsFullCacheType() throws Exception {
+ createDimTable(dimRecord(1L, "alice", "beijing"));
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders_full_cache_options", 1);
+
+ String joinSql =
+ lookupJoinSql(
+ "o.order_id, u.name",
+ "orders_full_cache_options",
+ "OPTIONS('lookup.cache'='FULL')",
+ null);
+
+ assertThat(SqlHelpers.sql(streamEnv, joinSql)).containsExactly(Row.of(1L,
"alice"));
+ }
+
+ @Test
+ void lookupJoinRejectsUnsupportedCacheType() throws Exception {
+ createDimTable(dimRecord(1L, "alice", "beijing"));
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders_unsupported_cache", 1);
+
+ String joinSql =
+ lookupJoinSql(
+ "o.order_id, u.name",
+ "orders_unsupported_cache",
+ "OPTIONS('lookup.cache'='PARTIAL')",
+ null);
+
+ assertThatThrownBy(() -> SqlHelpers.sql(streamEnv, joinSql))
+ .as("An Iceberg table cannot be point-looked-up, so PARTIAL must be
rejected")
+ .hasStackTraceContaining("only supports lookup.cache=FULL");
+ }
+
+ @Test
+ void lookupJoinLoadsCacheEagerlyByDefault() throws Exception {
+ String dataFileLocation = createEmptyProbeAndUnreadableDimTable();
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+
+ assertThatThrownBy(
+ () -> SqlHelpers.sql(streamEnv, lookupJoinSql("o.user_id, u.name",
"probe_empty")))
+ .as("The cache is loaded when the lookup function is opened, so the
job fails at startup")
+ .rootCause()
+ .hasMessageContaining(dataFileLocation);
+ }
+
+ @Test
+ void lookupJoinLoadsCacheLazilyWhenEagerLoadDisabled() throws Exception {
+ createEmptyProbeAndUnreadableDimTable();
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+
+ assertThat(
+ SqlHelpers.sql(
+ streamEnv,
+ lookupJoinSql(
+ "o.user_id, u.name",
+ "probe_empty",
+ "OPTIONS('lookup.full-cache.eager-load'='false')",
+ null)))
+ .as("The cache is loaded on the first lookup, and no probe row is ever
looked up")
+ .isEmpty();
+ }
+
+ @Test
+ void lookupJoinWithReorderedDimColumns() throws Exception {
+ createDimTable(dimRecord(1L, "alice", "beijing"), dimRecord(2L, "bob",
"shanghai"));
+
+ TableEnvironment streamEnv = getStreamingTableEnv();
+ createOrdersTable(streamEnv, "orders_reordered", 2);
+
+ String joinSql = lookupJoinSql("o.order_id, u.city, u.name",
"orders_reordered");
+
+ assertThat(SqlHelpers.sql(streamEnv, joinSql))
+ .containsExactlyInAnyOrder(Row.of(1L, "beijing", "alice"), Row.of(2L,
"shanghai", "bob"));
+ }
+
+ private static String lookupJoinSql(String selectClause, String probeTable) {
+ return lookupJoinSql(selectClause, probeTable, null, null);
+ }
+
+ private static String lookupJoinSql(
+ String selectClause, String probeTable, String options, String
extraCondition) {
+ return String.format(
+ "SELECT %s\n"
+ + "FROM %s AS o\n"
+ + "LEFT JOIN iceberg_catalog.`%s`.`%s`\n"
+ + "%s"
+ + " FOR SYSTEM_TIME AS OF o.proc_time AS u\n"
+ + " ON o.user_id = u.user_id%s",
+ selectClause,
+ probeTable,
+ TestFixtures.DATABASE,
+ TestFixtures.TABLE,
+ options == null ? "" : " /*+ " + options + " */\n",
+ extraCondition == null ? "" : " AND " + extraCondition);
+ }
+
+ private void createDefaultDimTable() throws IOException {
+ createDimTable(
+ dimRecord(1L, "alice", "beijing"),
+ dimRecord(2L, "bob", "shanghai"),
+ dimRecord(3L, "carol", "guangzhou"));
+ }
+
+ private void createDimTable(Record... records) throws IOException {
+ Table dimTable =
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER,
DIM_SCHEMA);
+ GenericAppenderHelper helper =
+ new GenericAppenderHelper(dimTable, FileFormat.PARQUET,
temporaryFolder);
+ DataFile dataFile = helper.writeFile(Lists.newArrayList(records));
+ helper.appendToTable(dataFile);
+ }
+
+ private static void createOrdersTable(
+ TableEnvironment tableEnvironment, String tableName, int numberOfRows) {
+ SqlHelpers.sql(
+ tableEnvironment,
+ "CREATE TEMPORARY TABLE %s (\n"
+ + " order_id BIGINT,\n"
+ + " user_id BIGINT,\n"
+ + " proc_time AS PROCTIME()\n"
+ + ") WITH (\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'number-of-rows' = '%d',\n"
+ + " 'fields.order_id.kind' = 'sequence',\n"
+ + " 'fields.order_id.start' = '1',\n"
+ + " 'fields.order_id.end' = '%d',\n"
+ + " 'fields.user_id.kind' = 'sequence',\n"
+ + " 'fields.user_id.start' = '1',\n"
+ + " 'fields.user_id.end' = '%d'\n"
+ + ")",
+ tableName,
+ numberOfRows,
+ numberOfRows,
+ numberOfRows);
+ }
+
+ private static Record dimRecord(long userId, String name, String city) {
+ Record record = GenericRecord.create(DIM_SCHEMA);
+ record.setField("user_id", userId);
+ record.setField("name", name);
+ record.setField("city", city);
+ return record;
+ }
+
+ private static void setUpTableEnv(TableEnvironment tableEnvironment) {
+ Configuration tableConf = tableEnvironment.getConfig().getConfiguration();
+ tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE,
true);
+
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM,
false);
+
+
tableEnvironment.getConfig().set("table.exec.resource.default-parallelism",
"1");
+ SqlHelpers.sql(
+ tableEnvironment,
+ "create catalog iceberg_catalog with ('type'='iceberg',
'catalog-type'='hadoop', 'warehouse'='%s')",
+ CATALOG_EXTENSION.warehouse());
+ SqlHelpers.sql(tableEnvironment, "use catalog iceberg_catalog");
+ }
+
+ private String createEmptyProbeAndUnreadableDimTable() throws IOException {
+ Table dimTable =
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER,
DIM_SCHEMA);
+ GenericAppenderHelper helper =
+ new GenericAppenderHelper(dimTable, FileFormat.PARQUET,
temporaryFolder);
+ DataFile dataFile = helper.writeFile(Lists.newArrayList(dimRecord(1L,
"alice", "beijing")));
+ helper.appendToTable(dataFile);
+
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(TableIdentifier.of(TestFixtures.DATABASE, "probe_empty"),
DIM_SCHEMA);
+
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "CREATE TEMPORARY TABLE probe_empty (\n"
+ + " user_id BIGINT,\n"
+ + " name STRING,\n"
+ + " city STRING,\n"
+ + " proc_time AS PROCTIME()\n"
+ + ") WITH (\n"
+ + " 'connector' = 'iceberg',\n"
+ + " 'catalog-name' = 'iceberg_catalog',\n"
+ + " 'catalog-type' = 'hadoop',\n"
+ + " 'warehouse' = '%s',\n"
+ + " 'catalog-database' = '%s',\n"
+ + " 'catalog-table' = 'probe_empty'\n"
+ + ")",
+ CATALOG_EXTENSION.warehouse(),
+ TestFixtures.DATABASE);
+
+ dimTable.io().deleteFile(dataFile.location());
+ return dataFile.location();
+ }
+}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
index 6cfc18868a..7d52662245 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
@@ -46,7 +46,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
/** Use the IcebergSource (FLIP-27) */
-public class TestIcebergSourceSql extends TestSqlBase {
+public class TestIcebergSourceSql extends TestSourceSqlBase {
private static final Schema SCHEMA_TS =
new Schema(
required(1, "t1", Types.TimestampType.withoutZone()),
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
similarity index 74%
copy from
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
copy to
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
index fe4c327310..61ae30bbe1 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
@@ -18,17 +18,11 @@
*/
package org.apache.iceberg.flink.source;
-import static org.apache.iceberg.flink.TestFixtures.DATABASE;
import static org.assertj.core.api.Assertions.assertThat;
-import java.io.IOException;
-import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.types.Row;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
@@ -38,60 +32,14 @@ import org.apache.iceberg.data.GenericAppenderHelper;
import org.apache.iceberg.data.RandomGenericData;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.flink.FlinkConfigOptions;
-import org.apache.iceberg.flink.HadoopCatalogExtension;
-import org.apache.iceberg.flink.MiniFlinkClusterExtension;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-import org.junit.jupiter.api.io.TempDir;
/** Test other more advanced usage of SQL. They don't need to run for every
file format. */
-public abstract class TestSqlBase {
- @RegisterExtension
- public static MiniClusterExtension miniClusterExtension =
- MiniFlinkClusterExtension.createWithClassloaderCheckDisabled();
-
- @RegisterExtension
- public static final HadoopCatalogExtension CATALOG_EXTENSION =
- new HadoopCatalogExtension(DATABASE, TestFixtures.TABLE);
-
- @TempDir protected Path temporaryFolder;
-
- private volatile TableEnvironment tEnv;
-
- private volatile TableEnvironment streamingTEnv;
-
- protected TableEnvironment getTableEnv() {
- if (tEnv == null) {
- synchronized (this) {
- if (tEnv == null) {
- this.tEnv =
-
TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
- }
- }
- }
- return tEnv;
- }
-
- protected TableEnvironment getStreamingTableEnv() {
- if (streamingTEnv == null) {
- synchronized (this) {
- if (streamingTEnv == null) {
- this.streamingTEnv =
-
TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());
- }
- }
- }
-
- return streamingTEnv;
- }
-
- @BeforeEach
- public abstract void before() throws IOException;
+public abstract class TestSourceSqlBase extends TestSqlBase {
@Test
public void testResiduals() throws Exception {
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
index fe4c327310..1ad17af2f3 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
@@ -19,37 +19,20 @@
package org.apache.iceberg.flink.source;
import static org.apache.iceberg.flink.TestFixtures.DATABASE;
-import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
import java.nio.file.Path;
-import java.util.List;
-import java.util.Map;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.apache.flink.types.Row;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.TestHelpers;
-import org.apache.iceberg.data.GenericAppenderHelper;
-import org.apache.iceberg.data.RandomGenericData;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.HadoopCatalogExtension;
import org.apache.iceberg.flink.MiniFlinkClusterExtension;
-import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.TestFixtures;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
-/** Test other more advanced usage of SQL. They don't need to run for every
file format. */
+/** Base class for SQL tests, providing a mini cluster, a Hadoop catalog and
table environments. */
public abstract class TestSqlBase {
@RegisterExtension
public static MiniClusterExtension miniClusterExtension =
@@ -92,82 +75,4 @@ public abstract class TestSqlBase {
@BeforeEach
public abstract void before() throws IOException;
-
- @Test
- public void testResiduals() throws Exception {
- Table table =
- CATALOG_EXTENSION
- .catalog()
- .createTable(TestFixtures.TABLE_IDENTIFIER, TestFixtures.SCHEMA,
TestFixtures.SPEC);
-
- List<Record> writeRecords =
RandomGenericData.generate(TestFixtures.SCHEMA, 2, 0L);
- writeRecords.get(0).set(1, 123L);
- writeRecords.get(0).set(2, "2020-03-20");
- writeRecords.get(1).set(1, 456L);
- writeRecords.get(1).set(2, "2020-03-20");
-
- GenericAppenderHelper helper =
- new GenericAppenderHelper(table, FileFormat.PARQUET, temporaryFolder);
-
- List<Record> expectedRecords = Lists.newArrayList();
- expectedRecords.add(writeRecords.get(0));
-
- DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0),
writeRecords);
- DataFile dataFile2 =
- helper.writeFile(
- TestHelpers.Row.of("2020-03-21", 0),
- RandomGenericData.generate(TestFixtures.SCHEMA, 2, 0L));
- helper.appendToTable(dataFile1, dataFile2);
-
- org.apache.iceberg.flink.TestHelpers.assertRecords(
- run(Maps.newHashMap(), "where dt='2020-03-20' and id=123", "*"),
- expectedRecords,
- TestFixtures.SCHEMA);
- }
-
- @Test
- public void testExposeLocality() throws Exception {
- Table table =
- CATALOG_EXTENSION
- .catalog()
- .createTable(TestFixtures.TABLE_IDENTIFIER, TestFixtures.SCHEMA,
TestFixtures.SPEC);
-
- TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
- List<Record> expectedRecords =
RandomGenericData.generate(TestFixtures.SCHEMA, 10, 0L);
- expectedRecords.forEach(expectedRecord -> expectedRecord.set(2,
"2020-03-20"));
-
- GenericAppenderHelper helper =
- new GenericAppenderHelper(table, FileFormat.PARQUET, temporaryFolder);
- DataFile dataFile = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0),
expectedRecords);
- helper.appendToTable(dataFile);
-
- // test sql api
- Configuration tableConf = getTableEnv().getConfig().getConfiguration();
-
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO,
false);
-
- List<Row> results = SqlHelpers.sql(getTableEnv(), "select * from t");
- org.apache.iceberg.flink.TestHelpers.assertRecords(
- results, expectedRecords, TestFixtures.SCHEMA);
-
- // test table api
-
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO,
true);
- FlinkSource.Builder builder =
FlinkSource.forRowData().tableLoader(tableLoader).table(table);
-
- // When running with CI or local, `localityEnabled` will be false even if
this configuration is
- // enabled
- assertThat(SourceUtil.isLocalityEnabled(table, tableConf, true))
- .as("Expose split locality info should be false.")
- .isFalse();
-
- results = run(Maps.newHashMap(), "where dt='2020-03-20'", "*");
- org.apache.iceberg.flink.TestHelpers.assertRecords(
- results, expectedRecords, TestFixtures.SCHEMA);
- }
-
- protected List<Row> run(
- Map<String, String> options, String sqlFilter, String...
sqlSelectedFields) {
- String select = String.join(",", sqlSelectedFields);
- String optionStr = SqlHelpers.sqlOptionsToString(options);
- return SqlHelpers.sql(getTableEnv(), "select %s from t %s %s", select,
optionStr, sqlFilter);
- }
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
new file mode 100644
index 0000000000..d13d89d17a
--- /dev/null
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.lookup;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestIcebergLookup {
+ private static final Schema SCHEMA =
+ new Schema(
+ optional(1, "id", Types.LongType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get()));
+
+ private static final String[] PROJECTED_COLUMNS = {"id", "data", "category"};
+ private static final RowType ROW_TYPE = FlinkSchemaUtil.convert(SCHEMA);
+ private static final int[] ID_KEY_INDEXES = {0};
+ private static final int[] ID_AND_CATEGORY_KEY_INDEXES = {0, 2};
+
+ @TempDir private Path temporaryFolder;
+
+ @RegisterExtension
+ protected static final HadoopCatalogExtension CATALOG_EXTENSION =
+ new HadoopCatalogExtension(TestFixtures.DATABASE, TestFixtures.TABLE);
+
+ private IcebergFullCachingLookupFunction lookupFunction;
+
+ @AfterEach
+ void after() throws Exception {
+ if (lookupFunction != null) {
+ lookupFunction.close();
+ }
+ }
+
+ @Test
+ void lookupReaderReadsWithBaseFilters() throws Exception {
+ Table table = createTableWithRecords();
+ IcebergLookupReader reader =
+ lookupReader(table, ImmutableList.of(Expressions.equal("category",
"B")), false);
+
+ List<List<Object>> rows = Lists.newArrayList();
+ try (CloseableIterable<RowData> rowData =
reader.read(IcebergLookupReader.CURRENT_SNAPSHOT)) {
+ for (RowData row : rowData) {
+ rows.add(
+ ImmutableList.of(
+ row.getLong(0), row.getString(1).toString(),
row.getString(2).toString()));
+ }
+ }
+
+ assertThat(rows).containsExactly(ImmutableList.of(2L, "bob", "B"));
+ }
+
+ @Test
+ void lookupReaderRespectsCaseSensitivity() throws Exception {
+ Table table = createTableWithRecords();
+ List<Expression> filters = ImmutableList.of(Expressions.equal("CATEGORY",
"B"));
+
+ IcebergLookupReader caseSensitiveReader = lookupReader(table, filters,
true);
+ assertThatThrownBy(() -> readIds(caseSensitiveReader,
IcebergLookupReader.CURRENT_SNAPSHOT))
+ .as("Case sensitive lookup should reject a filter that doesn't match
the column case")
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("CATEGORY");
+
+ assertThat(readIds(lookupReader(table, filters, false),
IcebergLookupReader.CURRENT_SNAPSHOT))
+ .containsExactly(2L);
+ }
+
+ @Test
+ void lookupReaderReadsPinnedSnapshot() throws Exception {
+ Table table = createTableWithRecords();
+ long pinnedSnapshot = table.currentSnapshot().snapshotId();
+
+ appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+ IcebergLookupReader reader = lookupReader(table, ImmutableList.of(),
false);
+
+ assertThat(readIds(reader, pinnedSnapshot))
+ .as("Pinned snapshot should not contain the appended row")
+ .containsExactlyInAnyOrder(1L, 2L, 3L);
+ assertThat(readIds(reader, IcebergLookupReader.CURRENT_SNAPSHOT))
+ .as("Current snapshot should contain the appended row")
+ .containsExactlyInAnyOrder(1L, 2L, 3L, 6L);
+ }
+
+ @Test
+ void lookupFunctionReturnsRowsFromCache() throws Exception {
+ Table table = createTableWithRecords();
+
+ lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+ lookupFunction.open(new FunctionContext(null));
+
+ assertThat(lookupFunction.lookup(keyRow(1L)))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+ appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+ assertThat(lookupFunction.lookup(keyRow(6L))).isEmpty();
+ assertThat(lookupFunction.lookup(keyRow(404L))).isEmpty();
+ assertThat(lookupFunction.lookup(keyRow(1L)))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+ }
+
+ @Test
+ void lookupFunctionWithMultiColumnKey() throws Exception {
+ Table table = createTableWithRecords();
+
+ lookupFunction = newLookupFunction(ID_AND_CATEGORY_KEY_INDEXES, false);
+ lookupFunction.open(new FunctionContext(null));
+
+ assertThat(lookupFunction.lookup(keyRow(1L, "A")))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+ assertThat(lookupFunction.lookup(keyRow(1L, "B"))).isEmpty();
+
+ appendRecords(table, ImmutableList.of(record(3L, "carol-2", "A")));
+
+ assertThat(lookupFunction.lookup(keyRow(3L, "A")))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 3L, "carol", "A"));
+ }
+
+ @Test
+ void lookupFunctionLoadsCacheLazily() throws Exception {
+ Table table = createTableWithRecords();
+
+ lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+ lookupFunction.open(new FunctionContext(null));
+
+ // Nothing is loaded yet, so the first lookup sees the row appended after
opening.
+ appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+ assertThat(lookupFunction.lookup(keyRow(6L)))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 6L, "frank", "D"));
+ }
+
+ @Test
+ void lookupFunctionLoadsCacheEagerly() throws Exception {
+ Table table = createTableWithRecords();
+
+ lookupFunction = newLookupFunction(ID_KEY_INDEXES, true);
+ lookupFunction.open(new FunctionContext(null));
+
+ appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+ assertThat(lookupFunction.lookup(keyRow(6L))).isEmpty();
+ assertThat(lookupFunction.lookup(keyRow(1L)))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+ }
+
+ @Test
+ void lookupFunctionHandlesEmptyTable() throws Exception {
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER,
SCHEMA);
+
+ lookupFunction = newLookupFunction(ID_KEY_INDEXES, true);
+ lookupFunction.open(new FunctionContext(null));
+
+ assertThat(lookupFunction.lookup(keyRow(1L))).isEmpty();
+ }
+
+ @Test
+ void lookupFunctionHandlesMultipleRowsPerKey() throws Exception {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA);
+ appendRecords(
+ table,
+ ImmutableList.of(
+ record(1L, "alice", "A"), record(1L, "alice-2", "A"), record(2L,
"bob", "B")));
+
+ lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+ lookupFunction.open(new FunctionContext(null));
+
+ assertThat(lookupFunction.lookup(keyRow(1L)))
+ .hasSize(2)
+ .allSatisfy(row -> assertThat(row.getLong(0)).isEqualTo(1L));
+ assertThat(lookupFunction.lookup(keyRow(2L)))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 2L, "bob", "B"));
+ }
+
+ @Test
+ void lookupFunctionAppliesPushedFilters() throws Exception {
+ createTableWithRecords();
+
+ lookupFunction =
+ new IcebergFullCachingLookupFunction(
+ CATALOG_EXTENSION.tableLoader().clone(),
+ ROW_TYPE,
+ ID_KEY_INDEXES,
+ ImmutableList.of(Expressions.equal("category", "B")),
+ true,
+ false);
+ lookupFunction.open(new FunctionContext(null));
+
+ assertThat(lookupFunction.lookup(keyRow(1L))).isEmpty();
+ assertThat(lookupFunction.lookup(keyRow(2L)))
+ .singleElement()
+ .satisfies(row -> assertRow(row, 2L, "bob", "B"));
+ }
+
+ private IcebergFullCachingLookupFunction newLookupFunction(
+ int[] lookupKeyIndexes, boolean eagerLoad) {
+ return new IcebergFullCachingLookupFunction(
+ CATALOG_EXTENSION.tableLoader().clone(),
+ ROW_TYPE,
+ lookupKeyIndexes,
+ ImmutableList.of(),
+ false,
+ eagerLoad);
+ }
+
+ private static IcebergLookupReader lookupReader(
+ Table table, List<Expression> filters, boolean caseSensitive) {
+ return new IcebergLookupReader(
+ table, SCHEMA.select(PROJECTED_COLUMNS), filters, caseSensitive, null);
+ }
+
+ private static List<Long> readIds(IcebergLookupReader reader, long
snapshotId)
+ throws IOException {
+ List<Long> ids = Lists.newArrayList();
+ try (CloseableIterable<RowData> rows = reader.read(snapshotId)) {
+ for (RowData row : rows) {
+ ids.add(row.isNullAt(0) ? null : row.getLong(0));
+ }
+ }
+
+ return ids;
+ }
+
+ private Table createTableWithRecords() throws Exception {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA);
+ appendRecords(
+ table,
+ ImmutableList.of(
+ record(1L, "alice", "A"), record(2L, "bob", "B"), record(3L,
"carol", "A")));
+ return table;
+ }
+
+ private void appendRecords(Table table, List<Record> records) throws
Exception {
+ new GenericAppenderHelper(table, FileFormat.PARQUET,
temporaryFolder).appendToTable(records);
+ }
+
+ private static GenericRecord record(Long id, String data, String category) {
+ GenericRecord record = GenericRecord.create(SCHEMA);
+ record.setField("id", id);
+ record.setField("data", data);
+ record.setField("category", category);
+ return record;
+ }
+
+ private static RowData keyRow(Long id) {
+ GenericRowData row = new GenericRowData(1);
+ row.setField(0, id);
+ return row;
+ }
+
+ private static RowData keyRow(Long id, String category) {
+ GenericRowData row = new GenericRowData(2);
+ row.setField(0, id);
+ row.setField(1, StringData.fromString(category));
+ return row;
+ }
+
+ private static void assertRow(RowData row, long id, String data, String
category) {
+ assertThat(row.getLong(0)).isEqualTo(id);
+ assertThat(row.getString(1)).hasToString(data);
+ assertThat(row.getString(2)).hasToString(category);
+ }
+}
