This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 4045158fa [core] Support timestamp type in Iceberg compatibility
(#4318)
4045158fa is described below
commit 4045158fa659484423476a960e68b61e7ecbbe81
Author: tsreaper <[email protected]>
AuthorDate: Tue Oct 15 11:52:40 2024 +0800
[core] Support timestamp type in Iceberg compatibility (#4318)
---
docs/content/migration/iceberg-compatibility.md | 22 ++++++
paimon-core/pom.xml | 1 -
.../iceberg/manifest/IcebergConversions.java | 28 +++++++
.../paimon/iceberg/metadata/IcebergDataField.java | 54 +++++++-------
.../paimon/iceberg/IcebergCompatibilityTest.java | 20 ++++-
paimon-flink/paimon-flink-1.17/pom.xml | 1 -
paimon-flink/paimon-flink-1.18/pom.xml | 1 -
paimon-flink/paimon-flink-1.19/pom.xml | 1 -
paimon-flink/paimon-flink-common/pom.xml | 13 ++++
.../flink/iceberg/FlinkIcebergITCaseBase.java | 87 ++++++++++++++++++++--
pom.xml | 1 +
11 files changed, 187 insertions(+), 42 deletions(-)
diff --git a/docs/content/migration/iceberg-compatibility.md
b/docs/content/migration/iceberg-compatibility.md
index 1373b6448..466f672a1 100644
--- a/docs/content/migration/iceberg-compatibility.md
+++ b/docs/content/migration/iceberg-compatibility.md
@@ -345,6 +345,28 @@ You can configure the following table option, so that
Paimon is forced to perfor
Note that full compaction is a resource-consuming process, so the value of
this table option should not be too small.
We recommend full compaction to be performed once or twice per hour.
+## Supported Types
+
+Paimon Iceberg compatibility currently supports the following data types.
+
+| Paimon Data Type | Iceberg Data Type |
+|----------------------------|-------------------|
+| `BOOLEAN` | `boolean` |
+| `INT` | `int` |
+| `BIGINT` | `long` |
+| `FLOAT` | `float` |
+| `DOUBLE` | `double` |
+| `DECIMAL` | `decimal` |
+| `CHAR` | `string` |
+| `VARCHAR` | `string` |
+| `BINARY` | `binary` |
+| `VARBINARY` | `binary` |
+| `DATE` | `date` |
+| `TIMESTAMP`* | `timestamp` |
+| `TIMESTAMP_LTZ`* | `timestamptz` |
+
+*: `TIMESTAMP` and `TIMESTAMP_LTZ` type only support precision from 4 to 6
+
## Other Related Table Options
<table class="table table-bordered">
diff --git a/paimon-core/pom.xml b/paimon-core/pom.xml
index a208d59b2..399f0b5d6 100644
--- a/paimon-core/pom.xml
+++ b/paimon-core/pom.xml
@@ -33,7 +33,6 @@ under the License.
<properties>
<frocksdbjni.version>6.20.3-ververica-2.0</frocksdbjni.version>
- <iceberg.version>1.6.1</iceberg.version>
</properties>
<dependencies>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergConversions.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergConversions.java
index 1d9e1c3b1..9048d46e4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergConversions.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergConversions.java
@@ -20,8 +20,12 @@ package org.apache.paimon.iceberg.manifest;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.utils.Preconditions;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
@@ -79,11 +83,26 @@ public class IcebergConversions {
case DECIMAL:
Decimal decimal = (Decimal) value;
return ByteBuffer.wrap((decimal.toUnscaledBytes()));
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return timestampToByteBuffer(
+ (Timestamp) value, ((TimestampType)
type).getPrecision());
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return timestampToByteBuffer(
+ (Timestamp) value, ((LocalZonedTimestampType)
type).getPrecision());
default:
throw new UnsupportedOperationException("Cannot serialize
type: " + type);
}
}
+ private static ByteBuffer timestampToByteBuffer(Timestamp timestamp, int
precision) {
+ Preconditions.checkArgument(
+ precision > 3 && precision <= 6,
+ "Paimon Iceberg compatibility only support timestamp type with
precision from 4 to 6.");
+ return ByteBuffer.allocate(8)
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .putLong(0, timestamp.toMicros());
+ }
+
public static Object toPaimonObject(DataType type, byte[] bytes) {
switch (type.getTypeRoot()) {
case BOOLEAN:
@@ -112,6 +131,15 @@ public class IcebergConversions {
DecimalType decimalType = (DecimalType) type;
return Decimal.fromUnscaledBytes(
bytes, decimalType.getPrecision(),
decimalType.getScale());
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ int timestampPrecision = ((TimestampType) type).getPrecision();
+ long timestampLong =
+
ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN).getLong();
+ Preconditions.checkArgument(
+ timestampPrecision > 3 && timestampPrecision <= 6,
+ "Paimon Iceberg compatibility only support timestamp
type with precision from 4 to 6.");
+ return Timestamp.fromMicros(timestampLong);
default:
throw new UnsupportedOperationException("Cannot deserialize
type: " + type);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
index 183fd5c00..c4505bbdd 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
@@ -20,8 +20,10 @@ package org.apache.paimon.iceberg.metadata;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
-import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.utils.Preconditions;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
@@ -57,6 +59,8 @@ public class IcebergDataField {
@JsonProperty(FIELD_TYPE)
private final String type;
+ @JsonIgnore private final DataType dataType;
+
@JsonProperty(FIELD_DOC)
private final String doc;
@@ -66,6 +70,7 @@ public class IcebergDataField {
dataField.name(),
!dataField.type().isNullable(),
toTypeString(dataField.type()),
+ dataField.type(),
dataField.description());
}
@@ -76,10 +81,16 @@ public class IcebergDataField {
@JsonProperty(FIELD_REQUIRED) boolean required,
@JsonProperty(FIELD_TYPE) String type,
@JsonProperty(FIELD_DOC) String doc) {
+ this(id, name, required, type, null, doc);
+ }
+
+ public IcebergDataField(
+ int id, String name, boolean required, String type, DataType
dataType, String doc) {
this.id = id;
this.name = name;
this.required = required;
this.type = type;
+ this.dataType = dataType;
this.doc = doc;
}
@@ -110,7 +121,7 @@ public class IcebergDataField {
@JsonIgnore
public DataType dataType() {
- return fromTypeString(type);
+ return Preconditions.checkNotNull(dataType);
}
private static String toTypeString(DataType dataType) {
@@ -137,38 +148,23 @@ public class IcebergDataField {
DecimalType decimalType = (DecimalType) dataType;
return String.format(
"decimal(%d, %d)", decimalType.getPrecision(),
decimalType.getScale());
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ int timestampPrecision = ((TimestampType)
dataType).getPrecision();
+ Preconditions.checkArgument(
+ timestampPrecision > 3 && timestampPrecision <= 6,
+ "Paimon Iceberg compatibility only support timestamp
type with precision from 4 to 6.");
+ return "timestamp";
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ int timestampLtzPrecision = ((LocalZonedTimestampType)
dataType).getPrecision();
+ Preconditions.checkArgument(
+ timestampLtzPrecision > 3 && timestampLtzPrecision <=
6,
+ "Paimon Iceberg compatibility only support timestamp
type with precision from 4 to 6.");
+ return "timestamptz";
default:
throw new UnsupportedOperationException("Unsupported data
type: " + dataType);
}
}
- private static DataType fromTypeString(String type) {
- if ("boolean".equals(type)) {
- return DataTypes.BOOLEAN();
- } else if ("int".equals(type)) {
- return DataTypes.INT();
- } else if ("long".equals(type)) {
- return DataTypes.BIGINT();
- } else if ("float".equals(type)) {
- return DataTypes.FLOAT();
- } else if ("double".equals(type)) {
- return DataTypes.DOUBLE();
- } else if ("date".equals(type)) {
- return DataTypes.DATE();
- } else if ("string".equals(type)) {
- return DataTypes.STRING();
- } else if ("binary".equals(type)) {
- return DataTypes.BYTES();
- } else if (type.startsWith("decimal")) {
- String[] precisionAndScale =
- type.substring("decimal(".length(), type.length() -
1).split(", ");
- return DataTypes.DECIMAL(
- Integer.parseInt(precisionAndScale[0]),
Integer.parseInt(precisionAndScale[1]));
- } else {
- throw new UnsupportedOperationException("Unsupported data type: "
+ type);
- }
- }
-
@Override
public int hashCode() {
return Objects.hash(id, name, required, type, doc);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
index dadbd557a..9a27d5618 100644
---
a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
@@ -26,6 +26,7 @@ import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
@@ -59,6 +60,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.math.BigDecimal;
import java.time.LocalDate;
+import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -374,7 +376,8 @@ public class IcebergCompatibilityTest {
DataTypes.STRING(),
DataTypes.BINARY(20),
DataTypes.VARBINARY(20),
- DataTypes.DATE()
+ DataTypes.DATE(),
+ DataTypes.TIMESTAMP(6)
},
new String[] {
"v_int",
@@ -387,7 +390,8 @@ public class IcebergCompatibilityTest {
"v_varchar",
"v_binary",
"v_varbinary",
- "v_date"
+ "v_date",
+ "v_timestamp"
});
FileStoreTable table =
createPaimonTable(rowType, Collections.emptyList(),
Collections.emptyList(), -1);
@@ -408,7 +412,8 @@ public class IcebergCompatibilityTest {
BinaryString.fromString("cat"),
"B_apple".getBytes(),
"B_cat".getBytes(),
- 100);
+ 100,
+ Timestamp.fromLocalDateTime(LocalDateTime.of(2024, 10,
10, 11, 22, 33)));
write.write(lowerBounds);
GenericRow upperBounds =
GenericRow.of(
@@ -422,7 +427,8 @@ public class IcebergCompatibilityTest {
BinaryString.fromString("dog"),
"B_banana".getBytes(),
"B_dog".getBytes(),
- 200);
+ 200,
+ Timestamp.fromLocalDateTime(LocalDateTime.of(2024, 10,
20, 11, 22, 33)));
write.write(upperBounds);
commit.commit(1, write.prepareCommit(false, 1));
@@ -450,6 +456,9 @@ public class IcebergCompatibilityTest {
} else if (type.getTypeRoot() == DataTypeRoot.DECIMAL) {
lower = new BigDecimal(lowerBounds.getField(i).toString());
upper = new BigDecimal(upperBounds.getField(i).toString());
+ } else if (type.getTypeRoot() ==
DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) {
+ lower = ((Timestamp) lowerBounds.getField(i)).toMicros();
+ upper = ((Timestamp) upperBounds.getField(i)).toMicros();
} else {
lower = lowerBounds.getField(i);
upper = upperBounds.getField(i);
@@ -460,6 +469,9 @@ public class IcebergCompatibilityTest {
if (type.getTypeRoot() == DataTypeRoot.DATE) {
expectedLower = LocalDate.ofEpochDay((int) lower).toString();
expectedUpper = LocalDate.ofEpochDay((int) upper).toString();
+ } else if (type.getTypeRoot() ==
DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) {
+ expectedLower = Timestamp.fromMicros((long) lower).toString();
+ expectedUpper = Timestamp.fromMicros((long) upper).toString();
}
assertThat(
diff --git a/paimon-flink/paimon-flink-1.17/pom.xml
b/paimon-flink/paimon-flink-1.17/pom.xml
index 06cc4f330..6e85d7101 100644
--- a/paimon-flink/paimon-flink-1.17/pom.xml
+++ b/paimon-flink/paimon-flink-1.17/pom.xml
@@ -36,7 +36,6 @@ under the License.
<properties>
<flink.version>1.17.2</flink.version>
<iceberg.flink.version>1.17</iceberg.flink.version>
- <iceberg.version>1.6.1</iceberg.version>
</properties>
<dependencies>
diff --git a/paimon-flink/paimon-flink-1.18/pom.xml
b/paimon-flink/paimon-flink-1.18/pom.xml
index add1915a5..d1c33f43f 100644
--- a/paimon-flink/paimon-flink-1.18/pom.xml
+++ b/paimon-flink/paimon-flink-1.18/pom.xml
@@ -36,7 +36,6 @@ under the License.
<properties>
<flink.version>1.18.1</flink.version>
<iceberg.flink.version>1.18</iceberg.flink.version>
- <iceberg.version>1.6.1</iceberg.version>
</properties>
<dependencies>
diff --git a/paimon-flink/paimon-flink-1.19/pom.xml
b/paimon-flink/paimon-flink-1.19/pom.xml
index 371ba54d1..736ac4c30 100644
--- a/paimon-flink/paimon-flink-1.19/pom.xml
+++ b/paimon-flink/paimon-flink-1.19/pom.xml
@@ -36,7 +36,6 @@ under the License.
<properties>
<flink.version>1.19.1</flink.version>
<iceberg.flink.version>1.19</iceberg.flink.version>
- <iceberg.version>1.6.1</iceberg.version>
</properties>
<dependencies>
diff --git a/paimon-flink/paimon-flink-common/pom.xml
b/paimon-flink/paimon-flink-common/pom.xml
index 3463aa9d9..4452af266 100644
--- a/paimon-flink/paimon-flink-common/pom.xml
+++ b/paimon-flink/paimon-flink-common/pom.xml
@@ -150,6 +150,19 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-core</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-data</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
index efa36020f..d76394c9e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
@@ -24,6 +24,13 @@ import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.io.CloseableIterable;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -192,7 +199,10 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ " v_decimal DECIMAL(8, 3),\n"
+ " v_varchar STRING,\n"
+ " v_varbinary VARBINARY(20),\n"
- + " v_date DATE\n"
+ + " v_date DATE,\n"
+ // it seems that Iceberg Flink connector has some bug
when filtering a
+ // timestamp_ltz, so we don't test it here
+ + " v_timestamp TIMESTAMP(6)\n"
+ ") PARTITIONED BY (pt) WITH (\n"
+ " 'metadata.iceberg.storage' = 'hadoop-catalog',\n"
+ " 'file.format' = '"
@@ -201,9 +211,9 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ ")");
tEnv.executeSql(
"INSERT INTO paimon.`default`.T VALUES "
- + "(1, 1, 1, true, 10, CAST(100.0 AS FLOAT),
1000.0, 123.456, 'cat', CAST('B_cat' AS VARBINARY(20)), DATE '2024-10-10'), "
- + "(2, 2, 2, false, 20, CAST(200.0 AS FLOAT),
2000.0, 234.567, 'dog', CAST('B_dog' AS VARBINARY(20)), DATE '2024-10-20'), "
- + "(3, 3, CAST(NULL AS INT), CAST(NULL AS
BOOLEAN), CAST(NULL AS BIGINT), CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE),
CAST(NULL AS DECIMAL(8, 3)), CAST(NULL AS STRING), CAST(NULL AS VARBINARY(20)),
CAST(NULL AS DATE))")
+ + "(1, 1, 1, true, 10, CAST(100.0 AS FLOAT),
1000.0, 123.456, 'cat', CAST('B_cat' AS VARBINARY(20)), DATE '2024-10-10',
TIMESTAMP '2024-10-10 11:22:33.123456'), "
+ + "(2, 2, 2, false, 20, CAST(200.0 AS FLOAT),
2000.0, 234.567, 'dog', CAST('B_dog' AS VARBINARY(20)), DATE '2024-10-20',
TIMESTAMP '2024-10-20 11:22:33.123456'), "
+ + "(3, 3, CAST(NULL AS INT), CAST(NULL AS
BOOLEAN), CAST(NULL AS BIGINT), CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE),
CAST(NULL AS DECIMAL(8, 3)), CAST(NULL AS STRING), CAST(NULL AS VARBINARY(20)),
CAST(NULL AS DATE), CAST(NULL AS TIMESTAMP(6)))")
.await();
tEnv.executeSql(
@@ -218,7 +228,8 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ " v_decimal DECIMAL(8, 3),\n"
+ " v_varchar STRING,\n"
+ " v_varbinary VARBINARY(20),\n"
- + " v_date DATE\n"
+ + " v_date DATE,\n"
+ + " v_timestamp TIMESTAMP(6)\n"
+ ") PARTITIONED BY (pt) WITH (\n"
+ " 'connector' = 'iceberg',\n"
+ " 'catalog-type' = 'hadoop',\n"
@@ -246,6 +257,11 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
.containsExactly(Row.of(1));
assertThat(collect(tEnv.executeSql("SELECT id FROM T where v_date =
'2024-10-10'")))
.containsExactly(Row.of(1));
+ assertThat(
+ collect(
+ tEnv.executeSql(
+ "SELECT id FROM T where v_timestamp =
TIMESTAMP '2024-10-10 11:22:33.123456'")))
+ .containsExactly(Row.of(1));
assertThat(collect(tEnv.executeSql("SELECT id FROM T where v_int IS
NULL")))
.containsExactly(Row.of(3));
assertThat(collect(tEnv.executeSql("SELECT id FROM T where v_boolean
IS NULL")))
@@ -264,6 +280,67 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
.containsExactly(Row.of(3));
assertThat(collect(tEnv.executeSql("SELECT id FROM T where v_date IS
NULL")))
.containsExactly(Row.of(3));
+ assertThat(collect(tEnv.executeSql("SELECT id FROM T where v_timestamp
IS NULL")))
+ .containsExactly(Row.of(3));
+ }
+
+ @ParameterizedTest
+ // orc writer does not write timestamp_ltz correctly, however we won't fix
it due to
+ // compatibility concern, so we don't test orc here
+ @ValueSource(strings = {"parquet"})
+ public void testFilterTimestampLtz(String format) throws Exception {
+ String warehouse = getTempDirPath();
+ TableEnvironment tEnv =
tableEnvironmentBuilder().batchMode().parallelism(2).build();
+ tEnv.executeSql(
+ "CREATE CATALOG paimon WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '"
+ + warehouse
+ + "'\n"
+ + ")");
+ tEnv.executeSql(
+ "CREATE TABLE paimon.`default`.T (\n"
+ + " id INT,"
+ + " v_timestampltz TIMESTAMP_LTZ(6)\n"
+ + ") WITH (\n"
+ + " 'metadata.iceberg.storage' = 'hadoop-catalog',\n"
+ + " 'file.format' = '"
+ + format
+ + "'\n"
+ + ")");
+ tEnv.executeSql(
+ "INSERT INTO paimon.`default`.T VALUES "
+ + "(1, CAST(TO_TIMESTAMP_LTZ(1100000000321, 3)
AS TIMESTAMP_LTZ(6))), "
+ + "(2, CAST(TO_TIMESTAMP_LTZ(1200000000321, 3)
AS TIMESTAMP_LTZ(6))), "
+ + "(3, CAST(NULL AS TIMESTAMP_LTZ(6)))")
+ .await();
+
+ HadoopCatalog icebergCatalog =
+ new HadoopCatalog(new Configuration(), warehouse + "/iceberg");
+ TableIdentifier icebergIdentifier = TableIdentifier.of("default", "T");
+ org.apache.iceberg.Table icebergTable =
icebergCatalog.loadTable(icebergIdentifier);
+
+ CloseableIterable<Record> result =
+ IcebergGenerics.read(icebergTable)
+ .where(Expressions.equal("v_timestampltz",
1100000000321000L))
+ .build();
+ List<Object> actual = new ArrayList<>();
+ for (Record record : result) {
+ actual.add(record.get(0));
+ }
+ result.close();
+ assertThat(actual).containsExactly(1);
+
+ result =
+ IcebergGenerics.read(icebergTable)
+ .where(Expressions.isNull("v_timestampltz"))
+ .build();
+ actual = new ArrayList<>();
+ for (Record record : result) {
+ actual.add(record.get(0));
+ }
+ result.close();
+ assertThat(actual).containsExactly(3);
}
@ParameterizedTest
diff --git a/pom.xml b/pom.xml
index 28952f46d..2d3f42b11 100644
--- a/pom.xml
+++ b/pom.xml
@@ -101,6 +101,7 @@ under the License.
<flink.forkCount>1C</flink.forkCount>
<flink.reuseForks>true</flink.reuseForks>
<testcontainers.version>1.19.1</testcontainers.version>
+ <iceberg.version>1.6.1</iceberg.version>
<!-- Can be set to any value to reproduce a specific build. -->
<test.randomization.seed/>
