You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by st...@apache.org on 2023/03/24 03:28:46 UTC
[iceberg] branch master updated: Flink: Backport #6614 to Flink 1.14 (#7166)
This is an automated email from the ASF dual-hosted git repository.
stevenwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 3fa9d672c7 Flink: Backport #6614 to Flink 1.14 (#7166)
3fa9d672c7 is described below
commit 3fa9d672c74a81464c0acfd3edecfe29925a4bf4
Author: xuzhiwen <10...@users.noreply.github.com>
AuthorDate: Fri Mar 24 11:28:37 2023 +0800
Flink: Backport #6614 to Flink 1.14 (#7166)
---
.../org/apache/iceberg/flink/CatalogLoader.java | 30 +++++++++++++++++++++-
.../java/org/apache/iceberg/flink/TableLoader.java | 18 ++++++++++++-
.../apache/iceberg/flink/source/IcebergSource.java | 2 +-
.../enumerator/ContinuousSplitPlannerImpl.java | 11 ++++++--
.../org/apache/iceberg/flink/TestTableLoader.java | 6 +++++
.../enumerator/TestContinuousSplitPlannerImpl.java | 26 +++++++++----------
6 files changed, 75 insertions(+), 18 deletions(-)
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/CatalogLoader.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/CatalogLoader.java
index 48fe7b8493..18473bf4f1 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/CatalogLoader.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/CatalogLoader.java
@@ -33,7 +33,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.rest.RESTCatalog;
/** Serializable loader to load an Iceberg {@link Catalog}. */
-public interface CatalogLoader extends Serializable {
+public interface CatalogLoader extends Serializable, Cloneable {
/**
* Create a new catalog with the provided properties. NOTICE: for flink, we may initialize the
@@ -45,6 +45,10 @@ public interface CatalogLoader extends Serializable {
*/
Catalog loadCatalog();
+ /** Clone a CatalogLoader. */
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ CatalogLoader clone();
+
static CatalogLoader hadoop(
String name, Configuration hadoopConf, Map<String, String> properties) {
return new HadoopCatalogLoader(name, hadoopConf, properties);
@@ -83,6 +87,12 @@ public interface CatalogLoader extends Serializable {
HadoopCatalog.class.getName(), catalogName, properties, hadoopConf.get());
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public CatalogLoader clone() {
+ return new HadoopCatalogLoader(catalogName, new Configuration(hadoopConf.get()), properties);
+ }
+
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
@@ -119,6 +129,12 @@ public interface CatalogLoader extends Serializable {
HiveCatalog.class.getName(), catalogName, properties, hadoopConf.get());
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public CatalogLoader clone() {
+ return new HiveCatalogLoader(catalogName, new Configuration(hadoopConf.get()), properties);
+ }
+
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
@@ -148,6 +164,12 @@ public interface CatalogLoader extends Serializable {
RESTCatalog.class.getName(), catalogName, properties, hadoopConf.get());
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public CatalogLoader clone() {
+ return new RESTCatalogLoader(catalogName, new Configuration(hadoopConf.get()), properties);
+ }
+
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
@@ -179,6 +201,12 @@ public interface CatalogLoader extends Serializable {
return CatalogUtil.loadCatalog(impl, name, properties, hadoopConf.get());
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public CatalogLoader clone() {
+ return new CustomCatalogLoader(name, properties, new Configuration(hadoopConf.get()), impl);
+ }
+
@Override
public String toString() {
return MoreObjects.toStringHelper(this).add("name", name).add("impl", impl).toString();
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java
index 1365b8b20c..6646338b62 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java
@@ -34,12 +34,16 @@ import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
* the cluster (for example, to get splits), not just on the client side. So we need an Iceberg
* table loader to get the {@link Table} object.
*/
-public interface TableLoader extends Closeable, Serializable {
+public interface TableLoader extends Closeable, Serializable, Cloneable {
void open();
Table loadTable();
+ /** Clone a TableLoader */
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ TableLoader clone();
+
static TableLoader fromCatalog(CatalogLoader catalogLoader, TableIdentifier identifier) {
return new CatalogTableLoader(catalogLoader, identifier);
}
@@ -77,6 +81,12 @@ public interface TableLoader extends Closeable, Serializable {
return tables.load(location);
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public TableLoader clone() {
+ return new HadoopTableLoader(location, new Configuration(hadoopConf.get()));
+ }
+
@Override
public void close() {}
@@ -111,6 +121,12 @@ public interface TableLoader extends Closeable, Serializable {
return catalog.loadTable(TableIdentifier.parse(identifier));
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public TableLoader clone() {
+ return new CatalogTableLoader(catalogLoader.clone(), TableIdentifier.parse(identifier));
+ }
+
@Override
public void close() throws IOException {
if (catalog instanceof Closeable) {
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
index df78bee6fa..d75e79abf3 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
@@ -187,7 +187,7 @@ public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEn
if (scanContext.isStreaming()) {
ContinuousSplitPlanner splitPlanner =
- new ContinuousSplitPlannerImpl(lazyTable(), scanContext, planningThreadName());
+ new ContinuousSplitPlannerImpl(tableLoader.clone(), scanContext, planningThreadName());
return new ContinuousIcebergEnumerator(
enumContext, assigner, scanContext, splitPlanner, enumState);
} else {
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java
index 904c47ce6a..85ecaaee1f 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java
@@ -28,6 +28,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.util.Preconditions;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.source.FlinkSplitPlanner;
import org.apache.iceberg.flink.source.ScanContext;
import org.apache.iceberg.flink.source.StreamingStartingStrategy;
@@ -46,13 +47,18 @@ public class ContinuousSplitPlannerImpl implements ContinuousSplitPlanner {
private final ScanContext scanContext;
private final boolean isSharedPool;
private final ExecutorService workerPool;
+ private final TableLoader tableLoader;
/**
+ * @param tableLoader A cloned tableLoader.
* @param threadName thread name prefix for worker pool to run the split planning. If null, a
* shared worker pool will be used.
*/
- public ContinuousSplitPlannerImpl(Table table, ScanContext scanContext, String threadName) {
- this.table = table;
+ public ContinuousSplitPlannerImpl(
+ TableLoader tableLoader, ScanContext scanContext, String threadName) {
+ this.tableLoader = tableLoader;
+ this.tableLoader.open();
+ this.table = tableLoader.loadTable();
this.scanContext = scanContext;
this.isSharedPool = threadName == null;
this.workerPool =
@@ -67,6 +73,7 @@ public class ContinuousSplitPlannerImpl implements ContinuousSplitPlanner {
if (!isSharedPool) {
workerPool.shutdown();
}
+ tableLoader.close();
}
@Override
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestTableLoader.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestTableLoader.java
index 61a821a9ac..93e97d5aa3 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestTableLoader.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestTableLoader.java
@@ -41,6 +41,12 @@ public class TestTableLoader implements TableLoader {
return TestTables.load(dir, "test");
}
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public TableLoader clone() {
+ return new TestTableLoader(dir.getAbsolutePath());
+ }
+
@Override
public void close() {}
}
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java
index e367dfaf29..fef61bc048 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java
@@ -116,7 +116,7 @@ public class TestContinuousSplitPlannerImpl {
.startingStrategy(StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTAL)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult emptyTableInitialDiscoveryResult = splitPlanner.planSplits(null);
Assert.assertTrue(emptyTableInitialDiscoveryResult.splits().isEmpty());
@@ -148,7 +148,7 @@ public class TestContinuousSplitPlannerImpl {
.startingStrategy(StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTAL)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
Assert.assertNull(initialResult.fromPosition());
@@ -181,7 +181,7 @@ public class TestContinuousSplitPlannerImpl {
.splitSize(1L)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult emptyTableInitialDiscoveryResult = splitPlanner.planSplits(null);
Assert.assertTrue(emptyTableInitialDiscoveryResult.splits().isEmpty());
@@ -219,7 +219,7 @@ public class TestContinuousSplitPlannerImpl {
.startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
Assert.assertNull(initialResult.fromPosition());
@@ -262,7 +262,7 @@ public class TestContinuousSplitPlannerImpl {
.startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult emptyTableInitialDiscoveryResult = splitPlanner.planSplits(null);
Assert.assertTrue(emptyTableInitialDiscoveryResult.splits().isEmpty());
@@ -294,7 +294,7 @@ public class TestContinuousSplitPlannerImpl {
.startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
Assert.assertNull(initialResult.fromPosition());
@@ -336,7 +336,7 @@ public class TestContinuousSplitPlannerImpl {
.build();
ContinuousSplitPlannerImpl splitPlanner =
new ContinuousSplitPlannerImpl(
- tableResource.table(), scanContextWithInvalidSnapshotId, null);
+ tableResource.tableLoader().clone(), scanContextWithInvalidSnapshotId, null);
AssertHelpers.assertThrows(
"Should detect invalid starting snapshot id",
@@ -364,7 +364,7 @@ public class TestContinuousSplitPlannerImpl {
ContinuousSplitPlannerImpl splitPlanner =
new ContinuousSplitPlannerImpl(
- tableResource.table(), scanContextWithInvalidSnapshotId, null);
+ tableResource.tableLoader().clone(), scanContextWithInvalidSnapshotId, null);
AssertHelpers.assertThrows(
"Should detect invalid starting snapshot id",
@@ -383,7 +383,7 @@ public class TestContinuousSplitPlannerImpl {
.startSnapshotId(snapshot2.snapshotId())
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
Assert.assertNull(initialResult.fromPosition());
@@ -428,7 +428,7 @@ public class TestContinuousSplitPlannerImpl {
.build();
ContinuousSplitPlannerImpl splitPlanner =
new ContinuousSplitPlannerImpl(
- tableResource.table(), scanContextWithInvalidSnapshotId, null);
+ tableResource.tableLoader().clone(), scanContextWithInvalidSnapshotId, null);
AssertHelpers.assertThrows(
"Should detect invalid starting snapshot timestamp",
@@ -451,7 +451,7 @@ public class TestContinuousSplitPlannerImpl {
ContinuousSplitPlannerImpl splitPlanner =
new ContinuousSplitPlannerImpl(
- tableResource.table(), scanContextWithInvalidSnapshotId, null);
+ tableResource.tableLoader().clone(), scanContextWithInvalidSnapshotId, null);
AssertHelpers.assertThrows(
"Should detect invalid starting snapshot timestamp",
@@ -470,7 +470,7 @@ public class TestContinuousSplitPlannerImpl {
.startSnapshotTimestamp(snapshot2.timestampMillis())
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
Assert.assertNull(initialResult.fromPosition());
@@ -520,7 +520,7 @@ public class TestContinuousSplitPlannerImpl {
.maxPlanningSnapshotCount(1)
.build();
ContinuousSplitPlannerImpl splitPlanner =
- new ContinuousSplitPlannerImpl(tableResource.table(), scanContext, null);
+ new ContinuousSplitPlannerImpl(tableResource.tableLoader().clone(), scanContext, null);
ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
Assert.assertNull(initialResult.fromPosition());