You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/12/18 04:34:55 UTC
[GitHub] [iceberg] stevenzwu opened a new pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
stevenzwu opened a new pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r554287700
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -76,7 +99,7 @@ protected TableEnvironment getTableEnv() {
TableResult tableResult = getTableEnv().executeSql(String.format(query, args));
tableResult.getJobClient().ifPresent(c -> {
try {
- c.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();
+ c.getJobExecutionResult().get();
Review comment:
BTW, the try-with-resource works well for Flink 1.12. It seems that it doesn't work with Flink 1.11 when I tried it in [the Flink test refactor PR](https://github.com/apache/iceberg/pull/2047)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r551681137
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -52,12 +53,25 @@ public TestFlinkScanSql(String fileFormat) {
@Override
public void before() throws IOException {
super.before();
- tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build());
- tEnv.executeSql(String.format(
+ getTableEnv().executeSql(String.format(
Review comment:
Why do we need to change here ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566046591
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -83,8 +83,8 @@ public void testGetTable() {
Types.NestedField.optional(1, "strV", Types.StringType.get())));
Assert.assertEquals(
Arrays.asList(
- TableColumn.of("id", DataTypes.BIGINT()),
- TableColumn.of("strV", DataTypes.STRING())),
+ TableColumn.physical("id", DataTypes.BIGINT()),
Review comment:
@stevenzwu I think we could change to use the loaded iceberg schema to accomplish the schema validation, so that we don't have to use the flink's TableSchema.
https://github.com/openinx/incubator-iceberg/commit/2a635719955f3eecd92685cb8f5486c46aa60095#diff-4c56ab08c19464dbe3351f71fc39345ee031a282b3e8dc1b107cbe9a1964d105R81
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r551679707
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -76,7 +99,7 @@ protected TableEnvironment getTableEnv() {
TableResult tableResult = getTableEnv().executeSql(String.format(query, args));
tableResult.getJobClient().ifPresent(c -> {
try {
- c.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();
+ c.getJobExecutionResult().get();
Review comment:
I read the javadoc about [TableResult](https://github.com/apache/flink/blob/97bfd049951f8d52a2e0aed14265074c4255ead0/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableResult.java#L186) again. The correct way to execute sql and get the results is:
```java
TableResult result = tEnv.execute("select ...");
// using try-with-resources statement
try (CloseableIterator<Row> it = result.collect()) {
it... // collect same data
}
```
Then I think we don't have to call `c.getJobExecutionResult().get()` here. How about removing the line101 ~ line 106 and the following part use `try-with-resources` statement ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-747881410
I think we should add the support for both flink 1.11 and flink 1.12 , Similar to spark2 and spark3.
If we upgrade flink 1.11 to flink 1.12, users who using flink 1.11 will not be able to use iceberg，what do you think ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu closed pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu closed pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568279327
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -136,13 +133,23 @@ public void testCreateTableIfNotExists() {
sql("DROP TABLE tl");
AssertHelpers.assertThrows("Table 'tl' should be dropped",
- NoSuchTableException.class, "Table does not exist: db.tl", () -> table("tl"));
+ NoSuchTableException.class,
+ "Table does not exist: " + getFullQualifiedTableName("tl"),
+ () -> table("tl"));
- sql("CREATE TABLE IF NO EXISTS tl(id BIGINT)");
+ sql("CREATE TABLE IF NOT EXISTS tl(id BIGINT)");
Assert.assertEquals(Maps.newHashMap(), table("tl").properties());
- sql("CREATE TABLE IF NOT EXISTS tl(id BIGINT) WITH ('location'='/tmp/location')");
- Assert.assertEquals("Should still be the old table.", Maps.newHashMap(), table("tl").properties());
+ final String uuid = UUID.randomUUID().toString();
Review comment:
Thanks a lot for the unit tests improvement !
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkCatalogTestBase.java
##########
@@ -119,6 +123,12 @@ protected String warehouseRoot() {
}
}
+ protected String getFullQualifiedTableName(String tableName) {
+ final List<String> levels = new ArrayList<>(Arrays.asList(icebergNamespace.levels()));
Review comment:
Nit: in apache iceberg, we usually use the unified `Lists.newArrayLists` to create a new ArrayList.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -107,7 +107,7 @@ public void testRenameTable() {
() -> getTableEnv().from("tl")
);
Assert.assertEquals(
- Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
+ Collections.singletonList(TableColumn.physical("id", DataTypes.BIGINT())),
Review comment:
@stevenzwu , do you think this issue need to fix in this PR ? I think we'd better to...
##########
File path: flink/src/test/java/org/apache/iceberg/flink/MiniClusterBase.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.TestBaseUtils;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * It will start a mini cluster with classloader.check-leaked-classloader=false, so that we won't break the unit tests
+ * because of the class loader leak issue. In our iceberg integration tests, there're some that will assert the results
+ * after finished the flink jobs, so actually we may access the class loader that has been closed by the flink task
+ * managers if we enable the switch classloader.check-leaked-classloader by default.
+ */
+public class MiniClusterBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ public static final Configuration CONFIG = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
Review comment:
I saw the [TestFlinkIcebergSinkV2](https://github.com/apache/iceberg/pull/1956/files#diff-13e2e5b52d0effe51e1b470df77cb08b5ec8cc4f3a7f0fd4e51ee212fc83f76aR73) also defines the similar mini cluster resources, then how about make it into a small methods so that the TestFlinkIcebergSinkV2 could reuse it ? For future defined mini cluster resource, we'd better also reuse this one because it will be easy to forget to disable this `CHECK_LEAKED_CLASSLOADER` switch for developers.
```java
@ClassRule
public static MiniClusterWithClientResource miniClusterResource = createMiniClusterResource();
@ClassRule
public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
public static MiniClusterWithClientResource createMiniClusterResource() {
return new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
.setConfiguration(CONFIG)
.build());
}
```
The TestFlinkIcebergSinkV2 could just use:
```java
@ClassRule
public static MiniClusterWithClientResource miniClusterResource = MiniClusterBase.createMiniClusterResource();
```
##########
File path: flink/src/main/java/org/apache/iceberg/flink/actions/Actions.java
##########
@@ -37,7 +43,7 @@ public static Actions forTable(StreamExecutionEnvironment env, Table table) {
}
public static Actions forTable(Table table) {
- return new Actions(StreamExecutionEnvironment.getExecutionEnvironment(), table);
+ return new Actions(StreamExecutionEnvironment.getExecutionEnvironment(CONFIG), table);
Review comment:
Thanks a lot for this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567808537
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
Seems all unit tests are passed https://github.com/openinx/incubator-iceberg/runs/1806279011 ! @stevenzwu , you could add this commit and then retry the travis CI again.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r546530572
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -91,7 +91,7 @@ public static Builder forRow(DataStream<Row> input, TableSchema tableSchema) {
DataType[] fieldDataTypes = tableSchema.getFieldDataTypes();
DataFormatConverters.RowConverter rowConverter = new DataFormatConverters.RowConverter(fieldDataTypes);
- return builderFor(input, rowConverter::toInternal, RowDataTypeInfo.of(rowType))
+ return builderFor(input, rowConverter::toInternal, InternalTypeInfo.of(rowType))
Review comment:
I checked the flink codebase, all the `RowData`'s TypeSerializer related classes are not expose to be `public`, they are marked as `Internal`. The only way I can think of is: Use a compactbility utility to load `InternalTypeInfo` or `RowDataTypeInfo` class if there's exist one.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-747881410
I think we should add the support for both flink 1.11 and flink 1.12 , Similar to spark2 and spark3.
If we upgrade flink 1.11 to flink 1.12, users who using flink 1.11 will not be able to use iceberg，what do you think
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-749540137
@stevenzwu , I think it's good time to upgrade flink from 1.11 to 1.12 after we've released the iceberg 0.11.0. About which way to accomplish the upgrade, I'm considering the same way as your proposed because of the flink compatibility issues. That means if someone just want the flink sink feature then it's better to choose iceberg 0.11.0 or 0.10.0, while if want more advanced features such as flink CDC or UPSERT stream, or the unified batch/stream iceberg source sink (the work you're working), then it's better to upgrade to use flink 1.12.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566823841
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
I think we need to disable the `CoreOptions.CHECK_LEAKED_CLASSLOADER` here ? Seems like we usually validation the iceberg tables results once we've terminated the flink job.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-749717064
@openinx I totally agree that we shouldn't upgrade flink to 1.12.0 until Iceberg 0.11.0 is released.
Regarding this PR, this is still some problem with the test failures `Trying to access closed classloader`. I tried to follow the same approach from what @JingsongLi did with [FLINK-19843]((https://issues.apache.org/jira/browse/FLINK-19843)). Still wasn't able to fix it yet.
Might be sth related to [Avro class cache as pointed out by Flink doc](https://ci.apache.org/projects/flink/flink-docs-stable/ops/debugging/debugging_classloading.html#unloading-of-dynamically-loaded-classes-in-user-code).
```
StdErr java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.
StdErr at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.ensureInner(FlinkUserCodeClassLoaders.java:161)
StdErr at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:168)
StdErr at java.base/java.lang.Class.forName0(Native Method)
StdErr at java.base/java.lang.Class.forName(Class.java:398)
StdErr at org.apache.iceberg.common.DynClasses$Builder.impl(DynClasses.java:68)
StdErr at org.apache.iceberg.avro.GenericAvroReader$ReadBuilder.record(GenericAvroReader.java:85)
StdErr at org.apache.iceberg.avro.GenericAvroReader$ReadBuilder.record(GenericAvroReader.java:72)
StdErr at org.apache.iceberg.avro.AvroSchemaVisitor.visit(AvroSchemaVisitor.java:50)
StdErr at org.apache.iceberg.avro.GenericAvroReader.initReader(GenericAvroReader.java:47)
StdErr at org.apache.iceberg.avro.GenericAvroReader.setSchema(GenericAvroReader.java:53)
StdErr at org.apache.iceberg.avro.ProjectionDatumReader.newDatumReader(ProjectionDatumReader.java:80)
StdErr at org.apache.iceberg.avro.ProjectionDatumReader.setSchema(ProjectionDatumReader.java:69)
StdErr at org.apache.avro.file.DataFileStream.initialize(DataFileStream.java:132)
StdErr at org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:106)
StdErr at org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:98)
StdErr at org.apache.avro.file.DataFileReader.openReader(DataFileReader.java:66)
StdErr at org.apache.iceberg.avro.AvroIterable.newFileReader(AvroIterable.java:100)
StdErr at org.apache.iceberg.avro.AvroIterable.getMetadata(AvroIterable.java:66)
StdErr at org.apache.iceberg.ManifestReader.<init>(ManifestReader.java:101)
StdErr at org.apache.iceberg.ManifestFiles.read(ManifestFiles.java:87)
StdErr at org.apache.iceberg.ManifestFiles.open(ManifestFiles.java:192)
StdErr at org.apache.iceberg.ManifestFiles.open(ManifestFiles.java:185)
StdErr at org.apache.iceberg.CatalogUtil.lambda$deleteFiles$4(CatalogUtil.java:106)
StdErr at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:404)
StdErr at org.apache.iceberg.util.Tasks$Builder.access$300(Tasks.java:70)
StdErr at org.apache.iceberg.util.Tasks$Builder$1.run(Tasks.java:310)
StdErr at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
StdErr at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
StdErr at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
StdErr at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
StdErr at java.base/java.lang.Thread.run(Thread.java:834)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-769800112
Skimmed this flink PR https://github.com/apache/flink/commit/fce82d7f56d5da3d3bf9ea6b66888d1350eb172f, sounds like we are still using the flink's class loader after we terminated the flink streaming job because we usually use the hadoop iceberg table to read records from it for assertion purpose.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-759692540
Totally agree that we should wait for Flink 1.12.1. There is also agreement that we should only upgrade Flink dep to 1.12 after Iceberg 0.11 release (probably this morning).
Yeah, if the connector lives in Flink, then we don't have the Flink versioning problem. Instead, we may have Iceberg versioning problem. Iceberg seems to be doing a fine job regarding API compatibility.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r554287700
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -76,7 +99,7 @@ protected TableEnvironment getTableEnv() {
TableResult tableResult = getTableEnv().executeSql(String.format(query, args));
tableResult.getJobClient().ifPresent(c -> {
try {
- c.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();
+ c.getJobExecutionResult().get();
Review comment:
BTW, removing the `JobClientgetJobExecutionResult().get()` works well for Flink 1.12. It seems that it doesn't work with Flink 1.11 when I tried it in [the Flink test refactor PR](https://github.com/apache/iceberg/pull/2047)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-754798065
Right now, CI test failed with `:iceberg-spark3-extensions:test` for java 8.
```
org.apache.iceberg.spark.extensions.TestCopyOnWriteDelete > testDeleteWithNotExistsSubquery[catalogName = spark_catalog, implementation = org.apache.iceberg.spark.SparkSessionCatalog, config = {type=hive, default-namespace=default, clients=1, parquet-enabled=false, cache-enabled=false}, format = avro, vectorized = false] FAILED
java.lang.RuntimeException: Failed to get table info from metastore default.table
Caused by:
org.apache.thrift.transport.TTransportException: java.net.SocketException: Broken pipe (Write failed)
Caused by:
java.net.SocketException: Broken pipe (Write failed)
```
I tried the same branch against the master branch in my own fork: https://github.com/stevenzwu/iceberg/pull/11. It also failed with connection in `:iceberg-hive3:compileJava`. I suspect maybe the build is taking a little longer and CI system aborted the build?
```
Execution failed for task ':iceberg-hive3:compileJava'.
> Could not resolve all files for configuration ':iceberg-hive3:compileClasspath'.
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.7.8.
Required by:
project :iceberg-hive3 > org.apache.hive:hive-exec:3.1.2 > org.apache.hadoop:hadoop-yarn-registry:3.1.0 > org.apache.hadoop:hadoop-yarn-common:3.1.0
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.7.8.
> Could not parse POM https://repo.maven.apache.org/maven2/com/fasterxml/jackson/jaxrs/jackson-jaxrs-json-provider/2.7.8/jackson-jaxrs-json-provider-2.7.8.pom
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-providers:2.7.8.
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-providers:2.7.8.
> Could not get resource 'https://repo.maven.apache.org/maven2/com/fasterxml/jackson/jaxrs/jackson-jaxrs-providers/2.7.8/jackson-jaxrs-providers-2.7.8.pom'.
> Could not GET 'https://repo.maven.apache.org/maven2/com/fasterxml/jackson/jaxrs/jackson-jaxrs-providers/2.7.8/jackson-jaxrs-providers-2.7.8.pom'.
> Connection reset
```
I have the same code in a diff branch. that PR is fine with CI build: https://github.com/stevenzwu/iceberg/pull/8.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-753566086
@tweise yeah. a few other people also asked for the same thing in [issue-1951](https://github.com/apache/iceberg/issues/1951).
I looked at the shim layer approach in `flink-connector-hive` module. It will probably work with existing code. But we will need to add an Iceberg source impl based on FLIP-27 source interface in 1.12. It is unclear how it works with the shim layer. New FLIP-27 Iceberg source code will extend from interfaces/classes available only in 1.12. I am not sure how it works unless we add a flink112 module in Iceberg.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r549934468
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSink.java
##########
@@ -46,21 +49,36 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Rule;
+import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
-public class TestFlinkIcebergSink extends AbstractTestBase {
+public class TestFlinkIcebergSink extends TestBaseUtils {
private static final TypeInformation<Row> ROW_TYPE_INFO = new RowTypeInfo(
SimpleDataUtil.FLINK_SCHEMA.getFieldTypes());
private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final org.apache.flink.configuration.Configuration config =
+ new org.apache.flink.configuration.Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
Review comment:
it seems that most Flink unit test with MiniCluster are using one taskmanager. I would stick with the convention.
On the other hand, this code is added to disable `CHECK_LEAKED_CLASSLOADER`. If we can fix that problem, we can stick with the `AbstractTestBase` base class
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566057769
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
##########
@@ -62,9 +61,8 @@ private TestHelpers() {
}
public static RowData copyRowData(RowData from, RowType rowType) {
- ExecutionConfig config = new ExecutionConfig();
TypeSerializer[] fieldSerializers = rowType.getChildren().stream()
- .map((LogicalType type) -> InternalSerializers.create(type, config))
+ .map((LogicalType type) -> InternalSerializers.create(type))
Review comment:
I'd prefer to create our iceberg's `RowDataCloner` (which is similar to the [RowDataConvert](https://github.com/apache/iceberg/blob/78e80d2a35c4c93e68776c31c24cc8ccb06fed4b/flink/src/test/java/org/apache/iceberg/flink/RowDataConverter.java#L49) ), so that we could get ride of this flink internal `InternalSerializers`.
Relying on flink internal interfaces that do not promise compatibility should be avoided as much as possible in subsequent development codes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568042540
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
@openinx thx a lot for identifying the missing integration tests that require disabling classloader check. I didn't realize the need to search for `StreamExecutionEnvironment.getExecutionEnvironment`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566064241
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -33,7 +33,7 @@
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.DataFormatConverters;
-import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
Review comment:
As this `InternalTypeInfo` is the flink internal API that we have to depends on in the flink+iceberg integration work, I'd like to move this `InternalTypeInfo.of(RowType rowType)` into a separate small utility methods, in case of changing every places that refer this internal class.
https://github.com/openinx/incubator-iceberg/commit/2a635719955f3eecd92685cb8f5486c46aa60095#diff-e4d1084ea160d030f3a225948d224678436b79eb5a94df230c71cdc7327f99d8R43
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566060147
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -52,12 +53,25 @@ public TestFlinkScanSql(String fileFormat) {
@Override
public void before() throws IOException {
super.before();
- tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build());
- tEnv.executeSql(String.format(
+ getTableEnv().executeSql(String.format(
Review comment:
OK, that sounds good to me.
Nit: here we could just use the `executeSQL` methods now. Maybe we could just rename the `executeSQL(String sql)` as `sql(String query, Object... args)`, so that we could have the same usage that is similar to the [FlinkTestBase#sql](https://github.com/apache/iceberg/blob/543a6cdd0538e505c4ab3fa0f0876a5a22bd7704/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java#L88).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567765847
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
There're still some ITCases which did not disable the `CHECK_LEAKED_CLASSLOADER` in current patch, so I made this commit: https://github.com/openinx/incubator-iceberg/commit/c8c2c8e8e72fd1b2051f1f946bb54c968e86c81d to re-trigger the travis CI again, let's see what will it say: https://github.com/openinx/incubator-iceberg/runs/1806279011
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566179182
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -617,12 +617,14 @@ public void testFilterNotPushDownLike() {
explainNoPushDown = getTableEnv().explainSql(sqlNoPushDown);
Assert.assertFalse("Explain should not contain FilterPushDown",
explainNoPushDown.contains(expectedFilterPushDownExplain));
+
sqlNoPushDown = "SELECT * FROM " + TABLE_NAME + " WHERE data LIKE '%%' ";
resultLike = sql(sqlNoPushDown);
- Assert.assertEquals("Should have 2 records", 2, resultLike.size());
+ Assert.assertEquals("Should have 3 records", 3, resultLike.size());
Review comment:
yeah. there are a couple of small behavior changes in 1.12. I also confirmed with zhangjun0x01
https://github.com/apache/iceberg/commit/98d9f6ad511b47b554ca14ae5d7fa8931161b3bf
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-749072124
I am going to close this PR, as it seems that we probably need more discussion on how to upgrade Flink from 1.11 to 1.12
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568285770
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkCatalogTestBase.java
##########
@@ -119,6 +123,12 @@ protected String warehouseRoot() {
}
}
+ protected String getFullQualifiedTableName(String tableName) {
+ final List<String> levels = new ArrayList<>(Arrays.asList(icebergNamespace.levels()));
Review comment:
will fix
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567017352
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
This is getting really tricky to debug. Can't reproduce it locally. Hence, can't set up break point or add debug logging to custom built Flink.
I tried to disable `CoreOptions.CHECK_LEAKED_CLASSLOADER` for all MiniClulster tests. That doesn't help.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r546534299
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -83,8 +83,8 @@ public void testGetTable() {
Types.NestedField.optional(1, "strV", Types.StringType.get())));
Assert.assertEquals(
Arrays.asList(
- TableColumn.of("id", DataTypes.BIGINT()),
- TableColumn.of("strV", DataTypes.STRING())),
+ TableColumn.physical("id", DataTypes.BIGINT()),
Review comment:
It's strange here, because I saw the `TableColumn` is marked as `PublicEvolving`, but after released flink 1.12.0 it did not have any Interface compatibility guarantee. At least, it should marked as `deprecated`, and keep it a major release.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx merged pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566220376
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -52,12 +53,25 @@ public TestFlinkScanSql(String fileFormat) {
@Override
public void before() throws IOException {
super.before();
- tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build());
- tEnv.executeSql(String.format(
+ getTableEnv().executeSql(String.format(
Review comment:
really like the suggestion. will update
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-754798065
Right now, CI test failed with `:iceberg-spark3-extensions:test` for java 8.
```
org.apache.iceberg.spark.extensions.TestCopyOnWriteDelete > testDeleteWithNotExistsSubquery[catalogName = spark_catalog, implementation = org.apache.iceberg.spark.SparkSessionCatalog, config = {type=hive, default-namespace=default, clients=1, parquet-enabled=false, cache-enabled=false}, format = avro, vectorized = false] FAILED
java.lang.RuntimeException: Failed to get table info from metastore default.table
Caused by:
org.apache.thrift.transport.TTransportException: java.net.SocketException: Broken pipe (Write failed)
Caused by:
java.net.SocketException: Broken pipe (Write failed)
```
I tried the same branch against the master branch in my own fork: https://github.com/stevenzwu/iceberg/pull/11. It also failed with connection in `:iceberg-hive3:compileJava`. I suspect maybe the build is taking a little longer and CI system aborted the build?
```
Execution failed for task ':iceberg-hive3:compileJava'.
> Could not resolve all files for configuration ':iceberg-hive3:compileClasspath'.
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.7.8.
Required by:
project :iceberg-hive3 > org.apache.hive:hive-exec:3.1.2 > org.apache.hadoop:hadoop-yarn-registry:3.1.0 > org.apache.hadoop:hadoop-yarn-common:3.1.0
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.7.8.
> Could not parse POM https://repo.maven.apache.org/maven2/com/fasterxml/jackson/jaxrs/jackson-jaxrs-json-provider/2.7.8/jackson-jaxrs-json-provider-2.7.8.pom
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-providers:2.7.8.
> Could not resolve com.fasterxml.jackson.jaxrs:jackson-jaxrs-providers:2.7.8.
> Could not get resource 'https://repo.maven.apache.org/maven2/com/fasterxml/jackson/jaxrs/jackson-jaxrs-providers/2.7.8/jackson-jaxrs-providers-2.7.8.pom'.
> Could not GET 'https://repo.maven.apache.org/maven2/com/fasterxml/jackson/jaxrs/jackson-jaxrs-providers/2.7.8/jackson-jaxrs-providers-2.7.8.pom'.
> Connection reset
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] tweise commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
tweise commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-759641805
@stevenzwu if Flink 1.12 is required to for the source implementation, then it should be fine to make that the minimum Flink version for the next release. Flink 1.12.1 voting is underway and the X.Y.1 version is typically what gets adopted downstream, X.Y.0 would be a bit premature for this purpose IMO.
Is the long term plan to contribute the connector to Flink? If so, the versioning problem goes away and maybe does not need to be addressed within Iceberg (depending how long it will take to contribute to Flink).
The Beam Flink version support approach has worked generally well. Even if only one of the versions is available in the IDE, that is normally sufficient for development.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-748171743
Are you suggesting that we have a module for each Flink minor version release (like 1.11 and 1.12)? I am worried that It can quickly grow out of control, as Flink does a minor version release typically every 4 months. Flink Kafka connector used to be in this model (like kafka08, kafka09, kafka10, ...). Now it moved to the universal connector and just track the latest release version of kafka-clients.
Or are you suggesting that the Flink connector code can handle all the 1.x Flink versions? I am not sure how flink hive connector handles some of the small API changes among all the versions.
As for spark2 and spark3, that is a major version upgrade. For upgrading Spark from 2.3 to 2.4, I assume Iceberg just upgrade the version in spark2 module. If in the future, Flink introduced major breaking API change and go up to 2.x, we probably should have a flink2 module in Iceberg.
Since the Flink Iceberg connector lives in the Iceberg project, I was thinking that the latest connector can just pick a Flink minor version as the paved path. That is why I was asking if we should wait for Flink 1.12.1 patch release for some bug fixes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r549475043
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSink.java
##########
@@ -46,21 +49,36 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Rule;
+import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
-public class TestFlinkIcebergSink extends AbstractTestBase {
+public class TestFlinkIcebergSink extends TestBaseUtils {
private static final TypeInformation<Row> ROW_TYPE_INFO = new RowTypeInfo(
SimpleDataUtil.FLINK_SCHEMA.getFieldTypes());
private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final org.apache.flink.configuration.Configuration config =
+ new org.apache.flink.configuration.Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
Review comment:
Would it make sense to use `2` here (or any value greater than 1)? At my work, we've found that some people's code runs into issues once it's distributed. I highly doubt that's likely to be the case here, given that typically those users were relatively new end users of flink, but it still might be helpful to run the tests with non-local shuffles (not in the same JVM). We could do 2 and then halve the number of slots per task manager.
Your call, but since this has been helpful for some people in my organization I thought I'd mention it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-749717064
@openinx I totally agree that we shouldn't upgrade flink to 1.12.0 until Iceberg 0.11.0 is released.
Regarding this PR, this is still some problem with the test failures `Trying to access closed classloader`. I tried to follow the same approach from what @JingsongLi did with [FLINK-19843]((https://issues.apache.org/jira/browse/FLINK-19843)). Still wasn't able to fix it yet.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx merged pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566925808
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
##########
@@ -62,9 +61,8 @@ private TestHelpers() {
}
public static RowData copyRowData(RowData from, RowType rowType) {
- ExecutionConfig config = new ExecutionConfig();
TypeSerializer[] fieldSerializers = rowType.getChildren().stream()
- .map((LogicalType type) -> InternalSerializers.create(type, config))
+ .map((LogicalType type) -> InternalSerializers.create(type))
Review comment:
Created issue-2184 to track this issue. probably a separate PR is better.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-771363814
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-771363814
@openinx I addressed your last round of comments. Thanks a lot for the great feedbacks.
thx for [catch the case](https://github.com/apache/iceberg/pull/1956#discussion_r568279087) about still using `TableColumn.physical`. I have fixed it. I also did a search and found no more usage.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r547273662
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -91,7 +91,7 @@ public static Builder forRow(DataStream<Row> input, TableSchema tableSchema) {
DataType[] fieldDataTypes = tableSchema.getFieldDataTypes();
DataFormatConverters.RowConverter rowConverter = new DataFormatConverters.RowConverter(fieldDataTypes);
- return builderFor(input, rowConverter::toInternal, RowDataTypeInfo.of(rowType))
+ return builderFor(input, rowConverter::toInternal, InternalTypeInfo.of(rowType))
Review comment:
Em, It's true.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566046736
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -107,7 +107,7 @@ public void testRenameTable() {
() -> getTableEnv().from("tl")
);
Assert.assertEquals(
- Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
+ Collections.singletonList(TableColumn.physical("id", DataTypes.BIGINT())),
Review comment:
ditto.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-771215176
@openinx I enabled `TestFlinkCatalogTable#testCreateTableIfNotExists()` test. I have to make two adjustments. In particular, I have to remove `WITH ('location'='/tmp/location')"` from the SQL in 2nd create. Otherwise, HadoopCatalog fails the validation. Also, for this test purpose, it is probably not important for the 2nd create to set a new location.
```
Test testCreateTableIfNotExists[catalogName = testhadoop baseNamespace = ](org.apache.iceberg.flink.TestFlinkCatalogTable) failed with:
org.apache.flink.table.api.TableException: Could not execute CreateTable in path `testhadoop`.`db`.`tl`
at org.apache.flink.table.catalog.CatalogManager.execute(CatalogManager.java:796)
at org.apache.flink.table.catalog.CatalogManager.createTable(CatalogManager.java:632)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:776)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)
at org.apache.iceberg.flink.FlinkTestBase.exec(FlinkTestBase.java:103)
at org.apache.iceberg.flink.FlinkTestBase.exec(FlinkTestBase.java:107)
at org.apache.iceberg.flink.FlinkTestBase.sql(FlinkTestBase.java:111)
at org.apache.iceberg.flink.TestFlinkCatalogTable.testCreateTableIfNotExists(TestFlinkCatalogTable.java:139)
...
Caused by: java.lang.IllegalArgumentException: Cannot set a custom location for a path-based table. Expected file:///var/folders/4d/3shff9g95pz81jmgrcn74zl00000gn/T/junit5877951167101792089/db/tl but got /tmp/location
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:142)
at org.apache.iceberg.hadoop.HadoopCatalog$HadoopCatalogTableBuilder.withLocation(HadoopCatalog.java:441)
at org.apache.iceberg.CachingCatalog$CachingTableBuilder.withLocation(CachingCatalog.java:191)
at org.apache.iceberg.CachingCatalog.createTable(CachingCatalog.java:104)
at org.apache.iceberg.flink.FlinkCatalog.createTable(FlinkCatalog.java:377)
at org.apache.flink.table.catalog.CatalogManager.lambda$createTable$10(CatalogManager.java:633)
at org.apache.flink.table.catalog.CatalogManager.execute(CatalogManager.java:790)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-771899750
@rdblue that is correct. master won't work with Flink 1.11 anymore.
The main reason is that the new FLIP-27 Flink source implementation will depend APIs only available from Flink 1.12. there were some discussions on how to make Flink connector works with both Flink 1.11 and 1.12 like adopting the gradle magic from Beam. In the end, it seems that most people are fine with having master only working with Flink 1.12.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566061811
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +36,28 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final org.apache.flink.configuration.Configuration config =
Review comment:
Nit: How about just import the `org.apache.flink.configuration.Configuration` class and use:
```java
private static final Configuration config = new Configuration().set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
```
here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-747881410
I think we should add the support for both flink 1.11 and flink 1.12 , similar to spark2 and spark3.
If we upgrade flink 1.11 to flink 1.12 directly, users who using flink 1.11 will not be able to use iceberg。
Similar to the support for hive in flink, flink also supports all hive versions ， what do you think ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566918036
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
It is already disabled here. We need to disable this check for other tests that are affected by this. CI tests were passing after disabling here. It started to fail again after I rebased with latest master.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566240925
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -33,7 +33,7 @@
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.DataFormatConverters;
-import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
Review comment:
will do assuming you meant just consolidation of calls to Flink Internal and PublicEvolve APIs. we don't need to support both Flink 1.11 and 1.12 in the compatibility util class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] twalthr commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567688796
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -83,8 +83,8 @@ public void testGetTable() {
Types.NestedField.optional(1, "strV", Types.StringType.get())));
Assert.assertEquals(
Arrays.asList(
- TableColumn.of("id", DataTypes.BIGINT()),
- TableColumn.of("strV", DataTypes.STRING())),
+ TableColumn.physical("id", DataTypes.BIGINT()),
Review comment:
sorry, this API break was a mistake and will be fixed in the next bugfix release
https://issues.apache.org/jira/browse/FLINK-21226
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566918036
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
It is already disabled here. We need to disable this check for other tests that are affected by this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-770826956
@stevenzwu , one more thing: we will need to enable this unit test https://github.com/apache/iceberg/blob/e58eba1fdd95c94b964e5b97322e73060d29d38a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java#L129 after we upgrade to flink 1.12.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566521593
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -617,12 +617,14 @@ public void testFilterNotPushDownLike() {
explainNoPushDown = getTableEnv().explainSql(sqlNoPushDown);
Assert.assertFalse("Explain should not contain FilterPushDown",
explainNoPushDown.contains(expectedFilterPushDownExplain));
+
sqlNoPushDown = "SELECT * FROM " + TABLE_NAME + " WHERE data LIKE '%%' ";
resultLike = sql(sqlNoPushDown);
- Assert.assertEquals("Should have 2 records", 2, resultLike.size());
+ Assert.assertEquals("Should have 3 records", 3, resultLike.size());
Review comment:
I tested it again. In hive 2.3.7, the sql `select * from mytest where data like '%' ` , data is null will not match. In flink 1.12, it will return the value of data is null. I have not found the related flink jira for this
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566047629
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -617,12 +617,14 @@ public void testFilterNotPushDownLike() {
explainNoPushDown = getTableEnv().explainSql(sqlNoPushDown);
Assert.assertFalse("Explain should not contain FilterPushDown",
explainNoPushDown.contains(expectedFilterPushDownExplain));
+
sqlNoPushDown = "SELECT * FROM " + TABLE_NAME + " WHERE data LIKE '%%' ";
resultLike = sql(sqlNoPushDown);
- Assert.assertEquals("Should have 2 records", 2, resultLike.size());
+ Assert.assertEquals("Should have 3 records", 3, resultLike.size());
Review comment:
Why do we need to change here ? in the new flink 1.12.1 release, the `null` data will be hit the `LIKE '%%'` ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-759692540
Totally agree that we should wait for Flink 1.12.1. There is also agreement that we should only upgrade Flink dep to 1.12 after Iceberg 0.11 release (probably this month).
Yeah, if the connector lives in Flink, then we don't have the Flink versioning problem. Instead, we may have Iceberg versioning problem. Iceberg seems to be doing a fine job regarding API compatibility.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566791767
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -107,7 +107,7 @@ public void testRenameTable() {
() -> getTableEnv().from("tl")
);
Assert.assertEquals(
- Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
+ Collections.singletonList(TableColumn.physical("id", DataTypes.BIGINT())),
Review comment:
Here, we could use the similar way to avoid to use TableColumn#physical. https://github.com/openinx/incubator-iceberg/commit/2a635719955f3eecd92685cb8f5486c46aa60095#diff-4c56ab08c19464dbe3351f71fc39345ee031a282b3e8dc1b107cbe9a1964d105R102
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
##########
@@ -62,9 +61,8 @@ private TestHelpers() {
}
public static RowData copyRowData(RowData from, RowType rowType) {
- ExecutionConfig config = new ExecutionConfig();
TypeSerializer[] fieldSerializers = rowType.getChildren().stream()
- .map((LogicalType type) -> InternalSerializers.create(type, config))
+ .map((LogicalType type) -> InternalSerializers.create(type))
Review comment:
Here we will use this to copy a `RowData` into a newly created `GenericRowData`, that means we don't have to use the complex `TypeSerializer` ( that is used for copying binary data for `BinaryRowData`). I think it will be good to remove the internal API if possible, so I'd recommend to use the similar `RowDataConverter` way to copy a totally new java objects for each fields.
BTW, I've discussed with flink sql team (@wuchong), the future apache flink will provides RowData copy utility to acomplish this RowData clone..
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
##########
@@ -62,9 +61,8 @@ private TestHelpers() {
}
public static RowData copyRowData(RowData from, RowType rowType) {
- ExecutionConfig config = new ExecutionConfig();
TypeSerializer[] fieldSerializers = rowType.getChildren().stream()
- .map((LogicalType type) -> InternalSerializers.create(type, config))
+ .map((LogicalType type) -> InternalSerializers.create(type))
Review comment:
Of course, we could open separate PR to address the `InternalSerializers` issue if you think it's necessary.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r551704679
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -76,7 +99,7 @@ protected TableEnvironment getTableEnv() {
TableResult tableResult = getTableEnv().executeSql(String.format(query, args));
tableResult.getJobClient().ifPresent(c -> {
try {
- c.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();
+ c.getJobExecutionResult().get();
Review comment:
I was wondering the same question regarding `c.getJobExecutionResult().get()`. Updated the PR per your suggestion.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -107,7 +121,14 @@ public void testResiduals() throws Exception {
}
private List<Row> executeSQL(String sql) {
- return Lists.newArrayList(tEnv.executeSql(sql).collect());
+ CloseableIterator<Row> iter = getTableEnv().executeSql(sql).collect();
Review comment:
updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r551683365
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -52,12 +53,25 @@ public TestFlinkScanSql(String fileFormat) {
@Override
public void before() throws IOException {
super.before();
- tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build());
- tEnv.executeSql(String.format(
+ getTableEnv().executeSql(String.format(
Review comment:
this is to avoid creating the table environment for each test method. it is the same pattern used by `FlinkTestBase`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-748261424
@JingsongLi saw the CI failure in Flink tests with `Trying to access closed classloader`, which I couldn't reproduce locally with `./gradlew :iceberg-flink:build`. It seems like the same problem as [FLINK-19843](https://issues.apache.org/jira/browse/FLINK-19843) that you fixed. I tried to close the result iterator as you did. But it doesn't seem to fix it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568284092
##########
File path: flink/src/test/java/org/apache/iceberg/flink/MiniClusterBase.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.TestBaseUtils;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * It will start a mini cluster with classloader.check-leaked-classloader=false, so that we won't break the unit tests
+ * because of the class loader leak issue. In our iceberg integration tests, there're some that will assert the results
+ * after finished the flink jobs, so actually we may access the class loader that has been closed by the flink task
+ * managers if we enable the switch classloader.check-leaked-classloader by default.
+ */
+public class MiniClusterBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ public static final Configuration CONFIG = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
Review comment:
I saw the [TestFlinkIcebergSinkV2](https://github.com/apache/iceberg/pull/1956/files#diff-13e2e5b52d0effe51e1b470df77cb08b5ec8cc4f3a7f0fd4e51ee212fc83f76aR73) also defines the similar mini cluster resources, then how about make it into a small methods so that the TestFlinkIcebergSinkV2 could reuse it ? For future defined mini cluster resource, we'd better also reuse this one because it will be easy to forget to disable this `CHECK_LEAKED_CLASSLOADER` switch for developers.
```java
@ClassRule
public static MiniClusterWithClientResource miniClusterResource = createMiniClusterResource();
@ClassRule
public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
public static MiniClusterWithClientResource createMiniClusterResource() {
return new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
.setConfiguration(CONFIG)
.build());
}
```
The TestFlinkIcebergSinkV2 could just use:
```java
@ClassRule
public static MiniClusterWithClientResource miniClusterResource = MiniClusterBase.createMiniClusterResource();
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566180136
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +36,28 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final org.apache.flink.configuration.Configuration config =
Review comment:
will fix
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568284500
##########
File path: flink/src/main/java/org/apache/iceberg/flink/actions/Actions.java
##########
@@ -37,7 +43,7 @@ public static Actions forTable(StreamExecutionEnvironment env, Table table) {
}
public static Actions forTable(Table table) {
- return new Actions(StreamExecutionEnvironment.getExecutionEnvironment(), table);
+ return new Actions(StreamExecutionEnvironment.getExecutionEnvironment(CONFIG), table);
Review comment:
Thanks a lot for this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568285770
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkCatalogTestBase.java
##########
@@ -119,6 +123,12 @@ protected String warehouseRoot() {
}
}
+ protected String getFullQualifiedTableName(String tableName) {
+ final List<String> levels = new ArrayList<>(Arrays.asList(icebergNamespace.levels()));
Review comment:
will fix
##########
File path: flink/src/test/java/org/apache/iceberg/flink/MiniClusterBase.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.TestBaseUtils;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * It will start a mini cluster with classloader.check-leaked-classloader=false, so that we won't break the unit tests
+ * because of the class loader leak issue. In our iceberg integration tests, there're some that will assert the results
+ * after finished the flink jobs, so actually we may access the class loader that has been closed by the flink task
+ * managers if we enable the switch classloader.check-leaked-classloader by default.
+ */
+public class MiniClusterBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ public static final Configuration CONFIG = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
Review comment:
That is a great idea. will update
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r546534299
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -83,8 +83,8 @@ public void testGetTable() {
Types.NestedField.optional(1, "strV", Types.StringType.get())));
Assert.assertEquals(
Arrays.asList(
- TableColumn.of("id", DataTypes.BIGINT()),
- TableColumn.of("strV", DataTypes.STRING())),
+ TableColumn.physical("id", DataTypes.BIGINT()),
Review comment:
It's strange here, because I saw the `TableColumn` is marked as `PublicEvolving`, but after released flink 1.12.0 it did not have any Interface compatibility guarantee. At least, it should marked as `deprecated`, and keep it a minor release.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-749717064
@openinx I totally agree that we shouldn't upgrade flink to 1.12.0 until Iceberg 0.11.0 is released.
Regarding this PR, this is still some problem with the test failures `Trying to access closed classloader`. I tried to follow the same approach from what @JingsongLi did with [FLINK-19843]((https://issues.apache.org/jira/browse/FLINK-19843)). Still wasn't able to fix it yet.
```
StdErr java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.
StdErr at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.ensureInner(FlinkUserCodeClassLoaders.java:161)
StdErr at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:168)
StdErr at java.base/java.lang.Class.forName0(Native Method)
StdErr at java.base/java.lang.Class.forName(Class.java:398)
StdErr at org.apache.iceberg.common.DynClasses$Builder.impl(DynClasses.java:68)
StdErr at org.apache.iceberg.avro.GenericAvroReader$ReadBuilder.record(GenericAvroReader.java:85)
StdErr at org.apache.iceberg.avro.GenericAvroReader$ReadBuilder.record(GenericAvroReader.java:72)
StdErr at org.apache.iceberg.avro.AvroSchemaVisitor.visit(AvroSchemaVisitor.java:50)
StdErr at org.apache.iceberg.avro.GenericAvroReader.initReader(GenericAvroReader.java:47)
StdErr at org.apache.iceberg.avro.GenericAvroReader.setSchema(GenericAvroReader.java:53)
StdErr at org.apache.iceberg.avro.ProjectionDatumReader.newDatumReader(ProjectionDatumReader.java:80)
StdErr at org.apache.iceberg.avro.ProjectionDatumReader.setSchema(ProjectionDatumReader.java:69)
StdErr at org.apache.avro.file.DataFileStream.initialize(DataFileStream.java:132)
StdErr at org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:106)
StdErr at org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:98)
StdErr at org.apache.avro.file.DataFileReader.openReader(DataFileReader.java:66)
StdErr at org.apache.iceberg.avro.AvroIterable.newFileReader(AvroIterable.java:100)
StdErr at org.apache.iceberg.avro.AvroIterable.getMetadata(AvroIterable.java:66)
StdErr at org.apache.iceberg.ManifestReader.<init>(ManifestReader.java:101)
StdErr at org.apache.iceberg.ManifestFiles.read(ManifestFiles.java:87)
StdErr at org.apache.iceberg.ManifestFiles.open(ManifestFiles.java:192)
StdErr at org.apache.iceberg.ManifestFiles.open(ManifestFiles.java:185)
StdErr at org.apache.iceberg.CatalogUtil.lambda$deleteFiles$4(CatalogUtil.java:106)
StdErr at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:404)
StdErr at org.apache.iceberg.util.Tasks$Builder.access$300(Tasks.java:70)
StdErr at org.apache.iceberg.util.Tasks$Builder$1.run(Tasks.java:310)
StdErr at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
StdErr at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
StdErr at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
StdErr at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
StdErr at java.base/java.lang.Thread.run(Thread.java:834)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-755770024
@zhangjun0x01 thx a lot for confirming your encounter of the problem. I also ran into the same problem locally when running Flink tests when I can tell my laptop became super slow. Maybe some tests are super heavy and cause high GC pause or sth.
I just rebased the branch with latest master. Now CI tests all pass.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-771885951
@openinx and @stevenzwu, what is the compatibility impact of this change? Does master no longer work with 1.11?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-771885951
@openinx and @stevenzwu, what is the compatibility impact of this change? Does master no longer work with 1.11?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566342579
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
##########
@@ -62,9 +61,8 @@ private TestHelpers() {
}
public static RowData copyRowData(RowData from, RowType rowType) {
- ExecutionConfig config = new ExecutionConfig();
TypeSerializer[] fieldSerializers = rowType.getChildren().stream()
- .map((LogicalType type) -> InternalSerializers.create(type, config))
+ .map((LogicalType type) -> InternalSerializers.create(type))
Review comment:
It will be great if we can achieve the goal. Looking at the behavior of clone, I am wondering if we will need to reinvent a lot of specific serializers (like `BooleanSerializer`) that `InternalSerializers` eventually depends on. Those type serializers are also internal just like `InternalSerializers`.
Ideally, it will be great if `RowData` interface provides a public `clone` API. Actually, in the FLIP-27 source, we need need to clone `RowData` since RowDataIterator returns a reused `RowData` object.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r546812762
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -91,7 +91,7 @@ public static Builder forRow(DataStream<Row> input, TableSchema tableSchema) {
DataType[] fieldDataTypes = tableSchema.getFieldDataTypes();
DataFormatConverters.RowConverter rowConverter = new DataFormatConverters.RowConverter(fieldDataTypes);
- return builderFor(input, rowConverter::toInternal, RowDataTypeInfo.of(rowType))
+ return builderFor(input, rowConverter::toInternal, InternalTypeInfo.of(rowType))
Review comment:
I think this is the shim layer approach that others suggested.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-755208338
I have also met several times, `iceberg-spark3-extensions` CI test failed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567766831
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
If the travis CI pass, then @stevenzwu, you can apply this commit into this PR to retry this travis CI:
```
curl -s https://github.com/openinx/incubator-iceberg/commit/c8c2c8e8e72fd1b2051f1f946bb54c968e86c81d.patch | git am
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568285911
##########
File path: flink/src/test/java/org/apache/iceberg/flink/MiniClusterBase.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.TestBaseUtils;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * It will start a mini cluster with classloader.check-leaked-classloader=false, so that we won't break the unit tests
+ * because of the class loader leak issue. In our iceberg integration tests, there're some that will assert the results
+ * after finished the flink jobs, so actually we may access the class loader that has been closed by the flink task
+ * managers if we enable the switch classloader.check-leaked-classloader by default.
+ */
+public class MiniClusterBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ public static final Configuration CONFIG = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
Review comment:
That is a great idea. will update
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568279327
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -136,13 +133,23 @@ public void testCreateTableIfNotExists() {
sql("DROP TABLE tl");
AssertHelpers.assertThrows("Table 'tl' should be dropped",
- NoSuchTableException.class, "Table does not exist: db.tl", () -> table("tl"));
+ NoSuchTableException.class,
+ "Table does not exist: " + getFullQualifiedTableName("tl"),
+ () -> table("tl"));
- sql("CREATE TABLE IF NO EXISTS tl(id BIGINT)");
+ sql("CREATE TABLE IF NOT EXISTS tl(id BIGINT)");
Assert.assertEquals(Maps.newHashMap(), table("tl").properties());
- sql("CREATE TABLE IF NOT EXISTS tl(id BIGINT) WITH ('location'='/tmp/location')");
- Assert.assertEquals("Should still be the old table.", Maps.newHashMap(), table("tl").properties());
+ final String uuid = UUID.randomUUID().toString();
Review comment:
Thanks a lot for the unit tests improvement !
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkCatalogTestBase.java
##########
@@ -119,6 +123,12 @@ protected String warehouseRoot() {
}
}
+ protected String getFullQualifiedTableName(String tableName) {
+ final List<String> levels = new ArrayList<>(Arrays.asList(icebergNamespace.levels()));
Review comment:
Nit: in apache iceberg, we usually use the unified `Lists.newArrayLists` to create a new ArrayList.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -107,7 +107,7 @@ public void testRenameTable() {
() -> getTableEnv().from("tl")
);
Assert.assertEquals(
- Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
+ Collections.singletonList(TableColumn.physical("id", DataTypes.BIGINT())),
Review comment:
@stevenzwu , do you think this issue need to fix in this PR ? I think we'd better to...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566925808
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
##########
@@ -62,9 +61,8 @@ private TestHelpers() {
}
public static RowData copyRowData(RowData from, RowType rowType) {
- ExecutionConfig config = new ExecutionConfig();
TypeSerializer[] fieldSerializers = rowType.getChildren().stream()
- .map((LogicalType type) -> InternalSerializers.create(type, config))
+ .map((LogicalType type) -> InternalSerializers.create(type))
Review comment:
Created [issue-2184](https://github.com/apache/iceberg/issues/2184) to track this issue. probably a separate PR is better.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r568042540
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
@openinx thx a lot for identifying the missing integration tests that require disabling classloader check. I didn't realize the need to search for `StreamExecutionEnvironment.getExecutionEnvironment`.
BTW, I also added `MiniClusterWithClientResource` to `TestFlinkIcebergSinkV2`, which might help speed up a tiny bit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-753566086
@tweise yeah. a few other people also asked for the same thing in [issue-1951](https://github.com/apache/iceberg/issues/1951).
I looked at the shim layer approach in `flink-connector-hive` module. It will work with the existing code (this PR). But we will need to add an Iceberg source impl based on FLIP-27 source interface in 1.12. It is unclear how it works with the shim layer. New FLIP-27 Iceberg source code will extend from interfaces/classes available only in 1.12. I am not sure how it works unless we add a flink112 module in Iceberg.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu edited a comment on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu edited a comment on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-753566086
@tweise yeah. a few other people also asked for the same thing in [issue-1951](https://github.com/apache/iceberg/issues/1951).
I looked at the shim layer approach in `flink-connector-hive` module. It will probably work with existing code. But we will need to add an Iceberg source impl based on FLIP-27 source interface in 1.12. It is unclear how it works with the shim layer. New FLIP-27 Iceberg source code will extend from interfaces/classes available only in 1.12.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567726646
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -83,8 +83,8 @@ public void testGetTable() {
Types.NestedField.optional(1, "strV", Types.StringType.get())));
Assert.assertEquals(
Arrays.asList(
- TableColumn.of("id", DataTypes.BIGINT()),
- TableColumn.of("strV", DataTypes.STRING())),
+ TableColumn.physical("id", DataTypes.BIGINT()),
Review comment:
It's OK, we've planned to just upgrade the flink 1.12 because we needs the newly introduced API from 1.12.0 to develop our unified iceberg flink source/sink. Thanks @twalthr for the work.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-762559543
@zhangjun0x01 it seems that Flink 1.12 upgrade might have changed the SQL behavior a little bit. I have to update the `TestFlinkTableSource` to match the new behavior. Can you please take a look and let me know if there is a better alternative?
https://github.com/apache/iceberg/pull/1956/commits/98d9f6ad511b47b554ca14ae5d7fa8931161b3bf
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567727299
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
OK, I'd see how to reproduce those class loader issues.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r567766831
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -34,8 +37,27 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
-public abstract class FlinkTestBase extends AbstractTestBase {
+public abstract class FlinkTestBase extends TestBaseUtils {
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static final Configuration config = new Configuration()
+ // disable classloader check as Avro may cache class/object in the serializers.
+ .set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
Review comment:
If the travis CI pass, then @stevenzwu, you can apply this commit in this patch to retry this travis CI:
```
curl -s https://github.com/openinx/incubator-iceberg/commit/c8c2c8e8e72fd1b2051f1f946bb54c968e86c81d.patch | git am
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r566228198
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -83,8 +83,8 @@ public void testGetTable() {
Types.NestedField.optional(1, "strV", Types.StringType.get())));
Assert.assertEquals(
Arrays.asList(
- TableColumn.of("id", DataTypes.BIGINT()),
- TableColumn.of("strV", DataTypes.STRING())),
+ TableColumn.physical("id", DataTypes.BIGINT()),
Review comment:
thx a lot for the suggestion. will update
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r551680528
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -107,7 +121,14 @@ public void testResiduals() throws Exception {
}
private List<Row> executeSQL(String sql) {
- return Lists.newArrayList(tEnv.executeSql(sql).collect());
+ CloseableIterator<Row> iter = getTableEnv().executeSql(sql).collect();
Review comment:
Similar issues here. ( see https://github.com/apache/iceberg/pull/1956/files#r551679707 )
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-759633809
I looked at the Beam's approach (gradle magic to copy version specific code + shim/compat layer) that @tweise suggested. It seems to be the most flexible and can work here. It does add some complexity though, mainly on the gradle setup initially. Anyone is more familiar with gradle likes to take it up?
FLIP-27 source implementation can live in the 1.12 sub module (with gradle trick to copying code during build). HiveShim approach in Flink probably won't work with the FLIP-27 source implementation, which requires Flink 1.12 dep.
@tweise any drawbacks of the Beam approach on supporting different Flink versions that we should be aware? like how does it work with IDE?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#discussion_r554287700
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -76,7 +99,7 @@ protected TableEnvironment getTableEnv() {
TableResult tableResult = getTableEnv().executeSql(String.format(query, args));
tableResult.getJobClient().ifPresent(c -> {
try {
- c.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();
+ c.getJobExecutionResult().get();
Review comment:
BTW, the try-with-resource works well for Flink 1.12. It seems that it doesn't work with Flink 1.11 when I tried it in [the Flink test refactor PR](https://github.com/apache/iceberg/pull/2047)
##########
File path: flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
##########
@@ -76,7 +99,7 @@ protected TableEnvironment getTableEnv() {
TableResult tableResult = getTableEnv().executeSql(String.format(query, args));
tableResult.getJobClient().ifPresent(c -> {
try {
- c.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();
+ c.getJobExecutionResult().get();
Review comment:
BTW, removing the `JobClientgetJobExecutionResult().get()` works well for Flink 1.12. It seems that it doesn't work with Flink 1.11 when I tried it in [the Flink test refactor PR](https://github.com/apache/iceberg/pull/2047)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] tweise commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
tweise commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-753555082
As the Flink Iceberg connector is evolving, it would probably be better to support multiple versions of Flink. Otherwise users will lack the Iceberg connector improvements if they are not able to follow a forced Flink upgrade. Apache Beam can serve as example how this can be accomplished with a relatively small shim layer: https://github.com/apache/beam/tree/master/runners/flink
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #1956: upgrade flink version from 1.11.0 to 1.12.0
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-753566086
@tweise yeah. a few other people also asked for the same thing in [issue-1951](https://github.com/apache/iceberg/issues/1951).
I looked at the shim layer approach in `flink-connector-hive` module. It will probably work with existing code. But we will need to add an Iceberg source impl based on FLIP-27 source interface in 1.12. It is unclear how it works with the shim layer. New FLIP-27 Iceberg source code will only compile with 1.12.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1956: Flink: Upgrade version from 1.11.0 to 1.12.1
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1956:
URL: https://github.com/apache/iceberg/pull/1956#issuecomment-769787140
Seems we still have the class loader issue ( from the travis ):
```java
org.apache.iceberg.flink.sink.TestTaskWriters > testRandomData[format = parquet, partitioned = true] FAILED
java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.ensureInner(FlinkUserCodeClassLoaders.java:164)
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.getResource(FlinkUserCodeClassLoaders.java:183)
at org.apache.hadoop.conf.Configuration.getResource(Configuration.java:2337)
at org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2538)
at org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2492)
at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2405)
at org.apache.hadoop.conf.Configuration.get(Configuration.java:981)
at org.apache.hadoop.conf.Configuration.getTrimmed(Configuration.java:1031)
at org.apache.hadoop.conf.Configuration.getInt(Configuration.java:1251)
at org.apache.hadoop.io.compress.GzipCodec.createInputStream(GzipCodec.java:154)
at org.apache.parquet.hadoop.CodecFactory$HeapBytesDecompressor.decompress(CodecFactory.java:109)
at org.apache.parquet.hadoop.ColumnChunkPageReadStore$ColumnChunkPageReader$1.visit(ColumnChunkPageReadStore.java:103)
at org.apache.parquet.hadoop.ColumnChunkPageReadStore$ColumnChunkPageReader$1.visit(ColumnChunkPageReadStore.java:99)
at org.apache.parquet.column.page.DataPageV1.accept(DataPageV1.java:120)
at org.apache.parquet.hadoop.ColumnChunkPageReadStore$ColumnChunkPageReader.readPage(ColumnChunkPageReadStore.java:99)
at org.apache.iceberg.parquet.BaseColumnIterator.advance(BaseColumnIterator.java:60)
at org.apache.iceberg.parquet.BaseColumnIterator.setPageSource(BaseColumnIterator.java:51)
at org.apache.iceberg.parquet.ParquetValueReaders$PrimitiveReader.setPageSource(ParquetValueReaders.java:185)
at org.apache.iceberg.parquet.ParquetValueReaders$OptionReader.setPageSource(ParquetValueReaders.java:369)
at org.apache.iceberg.parquet.ParquetValueReaders$StructReader.setPageSource(ParquetValueReaders.java:685)
at org.apache.iceberg.parquet.ParquetReader$FileIterator.advance(ParquetReader.java:142)
at org.apache.iceberg.parquet.ParquetReader$FileIterator.next(ParquetReader.java:112)
at org.apache.iceberg.io.CloseableIterable$ConcatCloseableIterable$ConcatCloseableIterator.next(CloseableIterable.java:206)
at org.apache.iceberg.relocated.com.google.common.collect.Iterators.addAll(Iterators.java:356)
at org.apache.iceberg.relocated.com.google.common.collect.Iterables.addAll(Iterables.java:320)
at org.apache.iceberg.relocated.com.google.common.collect.HashMultiset.create(HashMultiset.java:61)
at org.apache.iceberg.flink.SimpleDataUtil.assertTableRecords(SimpleDataUtil.java:194)
at org.apache.iceberg.flink.SimpleDataUtil.assertTableRecords(SimpleDataUtil.java:200)
at org.apache.iceberg.flink.SimpleDataUtil.assertTableRows(SimpleDataUtil.java:183)
at org.apache.iceberg.flink.sink.TestTaskWriters.testRandomData(TestTaskWriters.java:234)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org