[flink] branch master updated: [FLINK-13012][hive] Handle default partition name of Hive table
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1d81374 [FLINK-13012][hive] Handle default partition name of Hive
table
1d81374 is described below
commit 1d81374ce344e7bde410792c8a2efd6f4e707b9f
Author: Rui Li
AuthorDate: Thu Jul 11 21:08:22 2019 +0800
[FLINK-13012][hive] Handle default partition name of Hive table
This closes #9088
---
.../connectors/hive/HiveTableOutputFormat.java | 15 +++-
.../batch/connectors/hive/HiveTableSource.java | 11 ++-
.../connectors/hive/TableEnvHiveConnectorTest.java | 88 ++
3 files changed, 109 insertions(+), 5 deletions(-)
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
index acb1bf9..ade5830 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
@@ -141,6 +141,9 @@ public class HiveTableOutputFormat extends
HadoopOutputFormatCommonBase imp
// to convert Flink object to Hive object
private transient HiveObjectConversion[] hiveConversions;
+ // used when partition values is null or empty
+ private transient String defaultPartitionName;
+
public HiveTableOutputFormat(JobConf jobConf, ObjectPath tablePath,
CatalogTable table, HiveTablePartition hiveTablePartition,
Properties
tableProperties, boolean overwrite) {
super(jobConf.getCredentials());
@@ -298,6 +301,8 @@ public class HiveTableOutputFormat extends
HadoopOutputFormatCommonBase imp
rowObjectInspector =
ObjectInspectorFactory.getStandardStructObjectInspector(
Arrays.asList(rowTypeInfo.getFieldNames()).subList(0, rowTypeInfo.getArity() -
partitionColumns.size()),
objectInspectors);
+ defaultPartitionName =
jobConf.get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+
HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
}
}
@@ -310,10 +315,12 @@ public class HiveTableOutputFormat extends
HadoopOutputFormatCommonBase imp
// only need to check the dynamic partitions
final int numStaticPart =
hiveTablePartition.getPartitionSpec().size();
for (int i = dynamicPartitionOffset; i <
record.getArity(); i++) {
- // TODO: seems Hive also just calls
toString(), need further investigation to confirm
- // TODO: validate partition value
- String partVal =
record.getField(i).toString();
- dynPartSpec.put(partitionColumns.get(i
- dynamicPartitionOffset + numStaticPart), partVal);
+ Object field = record.getField(i);
+ String partitionValue = field != null ?
field.toString() : null;
+ if (partitionValue == null ||
partitionValue.isEmpty()) {
+ partitionValue =
defaultPartitionName;
+ }
+ dynPartSpec.put(partitionColumns.get(i
- dynamicPartitionOffset + numStaticPart), partitionValue);
}
String partName =
Warehouse.makePartPath(dynPartSpec);
partitionWriter =
partitionToWriter.get(partName);
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
index 706442d..e0734f4 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
@@ -157,6 +157,8 @@ public class HiveTableSource extends
InputFormatTableSource implements Part
String tableName = tablePath.getObjectName();
List partitionColNames =
catalogTable.getPartitionKeys();
if (partitionColNames != null &&
partitionColNames.size()
[flink] branch release-1.9 updated: [FLINK-13012][hive] Handle default partition name of Hive table
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 25e6143 [FLINK-13012][hive] Handle default partition name of Hive
table
25e6143 is described below
commit 25e6143a3c425c47851cbd69a41d68fe53f62f91
Author: Rui Li
AuthorDate: Thu Jul 11 21:08:22 2019 +0800
[FLINK-13012][hive] Handle default partition name of Hive table
This closes #9088
---
.../connectors/hive/HiveTableOutputFormat.java | 15 +++-
.../batch/connectors/hive/HiveTableSource.java | 11 ++-
.../connectors/hive/TableEnvHiveConnectorTest.java | 88 ++
3 files changed, 109 insertions(+), 5 deletions(-)
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
index acb1bf9..ade5830 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableOutputFormat.java
@@ -141,6 +141,9 @@ public class HiveTableOutputFormat extends
HadoopOutputFormatCommonBase imp
// to convert Flink object to Hive object
private transient HiveObjectConversion[] hiveConversions;
+ // used when partition values is null or empty
+ private transient String defaultPartitionName;
+
public HiveTableOutputFormat(JobConf jobConf, ObjectPath tablePath,
CatalogTable table, HiveTablePartition hiveTablePartition,
Properties
tableProperties, boolean overwrite) {
super(jobConf.getCredentials());
@@ -298,6 +301,8 @@ public class HiveTableOutputFormat extends
HadoopOutputFormatCommonBase imp
rowObjectInspector =
ObjectInspectorFactory.getStandardStructObjectInspector(
Arrays.asList(rowTypeInfo.getFieldNames()).subList(0, rowTypeInfo.getArity() -
partitionColumns.size()),
objectInspectors);
+ defaultPartitionName =
jobConf.get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+
HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
}
}
@@ -310,10 +315,12 @@ public class HiveTableOutputFormat extends
HadoopOutputFormatCommonBase imp
// only need to check the dynamic partitions
final int numStaticPart =
hiveTablePartition.getPartitionSpec().size();
for (int i = dynamicPartitionOffset; i <
record.getArity(); i++) {
- // TODO: seems Hive also just calls
toString(), need further investigation to confirm
- // TODO: validate partition value
- String partVal =
record.getField(i).toString();
- dynPartSpec.put(partitionColumns.get(i
- dynamicPartitionOffset + numStaticPart), partVal);
+ Object field = record.getField(i);
+ String partitionValue = field != null ?
field.toString() : null;
+ if (partitionValue == null ||
partitionValue.isEmpty()) {
+ partitionValue =
defaultPartitionName;
+ }
+ dynPartSpec.put(partitionColumns.get(i
- dynamicPartitionOffset + numStaticPart), partitionValue);
}
String partName =
Warehouse.makePartPath(dynPartSpec);
partitionWriter =
partitionToWriter.get(partName);
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
index 706442d..e0734f4 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/batch/connectors/hive/HiveTableSource.java
@@ -157,6 +157,8 @@ public class HiveTableSource extends
InputFormatTableSource implements Part
String tableName = tablePath.getObjectName();
List partitionColNames =
catalogTable.getPartitionKeys();
if (partitionColNames != null &&
partitionColNa
[flink] branch master updated: [FLINK-13451][tests] Remove use of Unsafe.defineClass() from CommonTestUtils
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c5b133e [FLINK-13451][tests] Remove use of Unsafe.defineClass() from
CommonTestUtils
c5b133e is described below
commit c5b133e6b13ea468763730e45f872b2370c2a206
Author: Stephan Ewen
AuthorDate: Sun Jul 28 17:16:17 2019 +0200
[FLINK-13451][tests] Remove use of Unsafe.defineClass() from CommonTestUtils
The method Unsafe.defineClass() is removed in Java 11. To support Java 11,
we rework the method
"CommonTestUtils.createClassNotInClassPath()" to use a different mechanism.
This commit now writes the class byte code out to a temporary file and
create a new URLClassLoader that
loads the class from that file. That solution is not a complete drop-in
replacement, because it cannot
add the class to an existing class loader, but can only create a new pair
of (classloader & new-class-in-that-classloader).
Because of that, the commit also adjusts the existing tests to work with
that new mechanism.
This closes #9251
---
.../kryo/KryoSerializerClassLoadingTest.java | 23 +--
.../runtime/kryo/KryoSerializerSnapshotTest.java | 20 +--
.../CheckpointSettingsSerializableTest.java| 7 +-
.../runtime/executiongraph/ErrorInfoTest.java | 6 +-
.../flink/runtime/state/JavaSerializerTest.java| 22 +--
.../flink/core/testutils/CommonTestUtils.java | 177 +++--
.../flink/core/testutils/CommonTestUtilsTest.java | 53 ++
7 files changed, 203 insertions(+), 105 deletions(-)
diff --git
a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerClassLoadingTest.java
b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerClassLoadingTest.java
index 9823e11..0d5e908 100644
---
a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerClassLoadingTest.java
+++
b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerClassLoadingTest.java
@@ -23,14 +23,11 @@ import
org.apache.flink.api.common.typeutils.SerializerTestBase;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.core.testutils.CommonTestUtils;
+
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import java.io.Serializable;
-import java.net.URL;
-import java.net.URLClassLoader;
-
import static org.junit.Assert.fail;
/**
@@ -39,13 +36,9 @@ import static org.junit.Assert.fail;
*/
public class KryoSerializerClassLoadingTest extends SerializerTestBase
{
- /** Class loader for the object that is not in the test class path */
- private static final ClassLoader CLASS_LOADER =
- new URLClassLoader(new URL[0],
KryoSerializerClassLoadingTest.class.getClassLoader());
-
- /** An object that is not in the test class path */
- private static final Serializable OBJECT_OUT_OF_CLASSPATH =
-
CommonTestUtils.createObjectForClassNotInClassPath(CLASS_LOADER);
+ /** Class loader and object that is not in the test class path. */
+ private static final CommonTestUtils.ObjectAndClassLoader
OUTSIDE_CLASS_LOADING =
+ CommonTestUtils.createObjectFromNewClassLoader();
//
@@ -54,7 +47,7 @@ public class KryoSerializerClassLoadingTest extends
SerializerTestBase {
@Before
public void setupClassLoader() {
originalClassLoader =
Thread.currentThread().getContextClassLoader();
- Thread.currentThread().setContextClassLoader(CLASS_LOADER);
+
Thread.currentThread().setContextClassLoader(OUTSIDE_CLASS_LOADING.getClassLoader());
}
@After
@@ -67,7 +60,7 @@ public class KryoSerializerClassLoadingTest extends
SerializerTestBase {
@Test
public void guardTestAssumptions() {
try {
-
Class.forName(OBJECT_OUT_OF_CLASSPATH.getClass().getName());
+
Class.forName(OUTSIDE_CLASS_LOADING.getObject().getClass().getName());
fail("This test's assumptions are broken");
}
catch (ClassNotFoundException ignored) {
@@ -98,11 +91,11 @@ public class KryoSerializerClassLoadingTest extends
SerializerTestBase {
new Integer(7),
// an object whose class is not on the classpath
- OBJECT_OUT_OF_CLASSPATH,
+ OUTSIDE_CLASS_LOADING.getObject(),
// an object whose class IS on the classpa
[flink] branch master updated: [FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps watermark strategy
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b670b11 [FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps
watermark strategy
b670b11 is described below
commit b670b11f303657aa8175ec933ee29b377cb9e087
Author: Timo Walther
AuthorDate: Fri Jul 26 15:14:42 2019 +0200
[FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps watermark
strategy
This closes #9241.
---
.../flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
index 725f534..1ddbc31 100644
---
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
+++
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
@@ -36,13 +36,14 @@ public final class BoundedOutOfOrderTimestamps extends
PeriodicWatermarkAssigner
private static final long serialVersionUID = 1L;
private final long delay;
- private long maxTimestamp = Long.MIN_VALUE + 1;
+ private long maxTimestamp;
/**
* @param delay The delay by which watermarks are behind the maximum
observed timestamp.
*/
public BoundedOutOfOrderTimestamps(long delay) {
this.delay = delay;
+ maxTimestamp = Long.MIN_VALUE + delay;
}
@Override
[flink] branch release-1.9 updated: [FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps watermark strategy
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 0224d9b [FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps
watermark strategy
0224d9b is described below
commit 0224d9bc0a773633943282f5268770a8063a87a2
Author: Timo Walther
AuthorDate: Fri Jul 26 15:14:42 2019 +0200
[FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps watermark
strategy
This closes #9241.
---
.../flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
index 725f534..1ddbc31 100644
---
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
+++
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/wmstrategies/BoundedOutOfOrderTimestamps.java
@@ -36,13 +36,14 @@ public final class BoundedOutOfOrderTimestamps extends
PeriodicWatermarkAssigner
private static final long serialVersionUID = 1L;
private final long delay;
- private long maxTimestamp = Long.MIN_VALUE + 1;
+ private long maxTimestamp;
/**
* @param delay The delay by which watermarks are behind the maximum
observed timestamp.
*/
public BoundedOutOfOrderTimestamps(long delay) {
this.delay = delay;
+ maxTimestamp = Long.MIN_VALUE + delay;
}
@Override
[flink] branch master updated (b670b11 -> 12118b9)
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from b670b11 [FLINK-13429][table-common] Fix BoundedOutOfOrderTimestamps watermark strategy add 12118b9 [FLINK-13387][WebUI] Fix log download for old UI No new revisions were added by this update. Summary of changes: .../old-version/partials/taskmanager/taskmanager.log.html | 4 ++-- .../old-version/partials/taskmanager/taskmanager.stdout.html | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-)
[flink] branch release-1.9 updated: [FLINK-13387][WebUI] Fix log download for old UI
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch release-1.9 in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/release-1.9 by this push: new 5077f18 [FLINK-13387][WebUI] Fix log download for old UI 5077f18 is described below commit 5077f186c5fc0644f06145bb5fa7796f54be8900 Author: Chesnay Schepler AuthorDate: Fri Jul 26 13:41:34 2019 +0200 [FLINK-13387][WebUI] Fix log download for old UI --- .../old-version/partials/taskmanager/taskmanager.log.html | 4 ++-- .../old-version/partials/taskmanager/taskmanager.stdout.html | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.log.html b/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.log.html index 54ed86a..112849d 100644 --- a/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.log.html +++ b/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.log.html @@ -24,7 +24,7 @@ limitations under the License. Task Manager Logs - + @@ -36,4 +36,4 @@ limitations under the License. - \ No newline at end of file + diff --git a/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.stdout.html b/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.stdout.html index 7acef7a..c8d8092 100644 --- a/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.stdout.html +++ b/flink-runtime-web/web-dashboard/old-version/partials/taskmanager/taskmanager.stdout.html @@ -24,7 +24,7 @@ limitations under the License. Task Manager Output - + @@ -36,4 +36,4 @@ limitations under the License. - \ No newline at end of file +
[flink] 01/03: [FLINK-13245][network] Fix the bug of file resource leak while canceling partition request
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 526236a4d537e78dbac4c575611ef52b602923d9
Author: Zhijiang
AuthorDate: Tue Jul 16 23:00:09 2019 +0800
[FLINK-13245][network] Fix the bug of file resource leak while canceling
partition request
On producer side the netty handler receives the CancelPartitionRequest for
releasing the SubpartitionView resource.
In previous implementation we try to find the corresponding view via
available queue in PartitionRequestQueue. But
in reality the view is not always available to stay in this queue, then the
view would never be released.
Furthermore the release of ResultPartition/ResultSubpartitions is based on
the reference counter in ReleaseOnConsumptionResultPartition,
but while handling the CancelPartitionRequest in PartitionRequestQueue, the
ReleaseOnConsumptionResultPartition is never
notified of consumed subpartition. That means the reference counter would
never decrease to 0 to trigger partition release,
which would bring file resource leak in the case of
BoundedBlockingSubpartition.
In order to fix above two issues, the corresponding view is released via
all reader queue instead, and then it would call
ReleaseOnConsumptionResultPartition#onConsumedSubpartition meanwhile to
solve this bug.
---
.../io/network/netty/PartitionRequestQueue.java| 33
.../network/netty/CancelPartitionRequestTest.java | 4 +-
.../network/netty/PartitionRequestQueueTest.java | 92 ++
.../io/network/partition/PartitionTestUtils.java | 8 ++
4 files changed, 118 insertions(+), 19 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index f82a42f..b492ea6 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -138,9 +138,7 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
}
for (NetworkSequenceViewReader reader : allReaders.values()) {
- reader.notifySubpartitionConsumed();
- reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
+ releaseViewReader(reader);
}
allReaders.clear();
}
@@ -181,19 +179,14 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
return;
}
- // Cancel the request for the input channel
- int size = availableReaders.size();
- for (int i = 0; i < size; i++) {
- NetworkSequenceViewReader reader =
pollAvailableReader();
- if (reader.getReceiverId().equals(toCancel)) {
- reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
- } else {
- registerAvailableReader(reader);
- }
- }
+ // remove reader from queue of available readers
+ availableReaders.removeIf(reader ->
reader.getReceiverId().equals(toCancel));
- allReaders.remove(toCancel);
+ // remove reader from queue of all readers and release
its resource
+ final NetworkSequenceViewReader toRelease =
allReaders.remove(toCancel);
+ if (toRelease != null) {
+ releaseViewReader(toRelease);
+ }
} else {
ctx.fireUserEventTriggered(msg);
}
@@ -308,14 +301,20 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
private void releaseAllResources() throws IOException {
// note: this is only ever executed by one thread: the Netty IO
thread!
for (NetworkSequenceViewReader reader : allReaders.values()) {
- reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
+ releaseViewReader(reader);
}
availableReaders.clear();
allReaders.clear();
}
+ private void releaseViewReader(NetworkSequenceViewReader reader) throws
IOException {
+ reader.notifySubpartitionConsumed();
+ reader.setRegisteredAsAvailable(f
[flink] branch release-1.9 updated (5077f18 -> 7116ab7)
This is an automated email from the ASF dual-hosted git repository. sewen pushed a change to branch release-1.9 in repository https://gitbox.apache.org/repos/asf/flink.git. from 5077f18 [FLINK-13387][WebUI] Fix log download for old UI new 526236a [FLINK-13245][network] Fix the bug of file resource leak while canceling partition request new f738c4e [FLINK-13245][network] Make subpartition consumption notification independant new 7116ab7 [FLINK-13245][network] Remove redundant bookkeeping for already canceled input channel IDs The 3 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../io/network/netty/PartitionRequestQueue.java| 43 +++--- .../ReleaseOnConsumptionResultPartition.java | 42 +++--- .../network/netty/CancelPartitionRequestTest.java | 4 +- .../network/netty/PartitionRequestQueueTest.java | 92 ++ .../io/network/partition/PartitionTestUtils.java | 8 ++ .../ReleaseOnConsumptionResultPartitionTest.java | 41 +- 6 files changed, 182 insertions(+), 48 deletions(-)
[flink] 03/03: [FLINK-13245][network] Remove redundant bookkeeping for already canceled input channel IDs
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7116ab71edc183d34d128453e06a3efc15ad8905
Author: Zhijiang
AuthorDate: Fri Jul 26 11:50:55 2019 +0200
[FLINK-13245][network] Remove redundant bookkeeping for already canceled
input channel IDs
---
.../runtime/io/network/netty/PartitionRequestQueue.java | 16
1 file changed, 16 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index b492ea6..4a845d1 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -26,7 +26,6 @@ import
org.apache.flink.runtime.io.network.partition.ProducerFailedException;
import
org.apache.flink.runtime.io.network.partition.consumer.InputChannel.BufferAndAvailability;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
-import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
@@ -40,7 +39,6 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayDeque;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -62,8 +60,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
/** All the readers created for the consumers' partition requests. */
private final ConcurrentMap
allReaders = new ConcurrentHashMap<>();
- private final Set released = Sets.newHashSet();
-
private boolean fatalError;
private ChannelHandlerContext ctx;
@@ -175,9 +171,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
} else if (msg.getClass() == InputChannelID.class) {
// Release partition view that get a cancel request.
InputChannelID toCancel = (InputChannelID) msg;
- if (released.contains(toCancel)) {
- return;
- }
// remove reader from queue of available readers
availableReaders.removeIf(reader ->
reader.getReceiverId().equals(toCancel));
@@ -222,7 +215,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
if (!reader.isReleased()) {
continue;
}
- markAsReleased(reader.getReceiverId());
Throwable cause =
reader.getFailureCause();
if (cause != null) {
@@ -312,14 +304,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
reader.notifySubpartitionConsumed();
reader.setRegisteredAsAvailable(false);
reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
- }
-
- /**
-* Marks a receiver as released.
-*/
- private void markAsReleased(InputChannelID receiverId) {
- released.add(receiverId);
}
// This listener is called after an element of the current
nonEmptyReader has been
[flink] 02/03: [FLINK-13245][network] Make subpartition consumption notification independant
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f738c4edd5858891236a41fc1e453b06cf815089
Author: Zhijiang
AuthorDate: Fri Jul 26 11:25:40 2019 +0200
[FLINK-13245][network] Make subpartition consumption notification
independant
---
.../ReleaseOnConsumptionResultPartition.java | 42 +++---
.../ReleaseOnConsumptionResultPartitionTest.java | 41 ++---
2 files changed, 67 insertions(+), 16 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
index 19ec681..766f500 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
@@ -22,7 +22,6 @@ import
org.apache.flink.runtime.io.network.buffer.BufferPoolOwner;
import org.apache.flink.util.function.FunctionWithException;
import java.io.IOException;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.flink.util.Preconditions.checkState;
@@ -31,11 +30,18 @@ import static
org.apache.flink.util.Preconditions.checkState;
*/
public class ReleaseOnConsumptionResultPartition extends ResultPartition {
+ private static final Object lock = new Object();
+
+ /**
+* A flag for each subpartition indicating whether it was already
consumed or not.
+*/
+ private final boolean[] consumedSubpartitions;
+
/**
* The total number of references to subpartitions of this result. The
result partition can be
* safely released, iff the reference count is zero.
*/
- private final AtomicInteger pendingReferences = new AtomicInteger();
+ private int numUnconsumedSubpartitions;
ReleaseOnConsumptionResultPartition(
String owningTaskName,
@@ -47,12 +53,13 @@ public class ReleaseOnConsumptionResultPartition extends
ResultPartition {
FunctionWithException bufferPoolFactory) {
super(owningTaskName, partitionId, partitionType,
subpartitions, numTargetKeyGroups, partitionManager, bufferPoolFactory);
- pendingReferences.set(subpartitions.length);
+ this.consumedSubpartitions = new boolean[subpartitions.length];
+ this.numUnconsumedSubpartitions = subpartitions.length;
}
@Override
public ResultSubpartitionView createSubpartitionView(int index,
BufferAvailabilityListener availabilityListener) throws IOException {
- checkState(pendingReferences.get() > 0, "Partition not
pinned.");
+ checkState(numUnconsumedSubpartitions > 0, "Partition not
pinned.");
return super.createSubpartitionView(index,
availabilityListener);
}
@@ -63,22 +70,33 @@ public class ReleaseOnConsumptionResultPartition extends
ResultPartition {
return;
}
- int refCnt = pendingReferences.decrementAndGet();
+ final int remainingUnconsumed;
- if (refCnt == 0) {
- partitionManager.onConsumedPartition(this);
- } else if (refCnt < 0) {
- throw new IllegalStateException("All references
released.");
+ // we synchronize only the bookkeeping section, to avoid
holding the lock during any
+ // calls into other components
+ synchronized (lock) {
+ if (consumedSubpartitions[subpartitionIndex]) {
+ // repeated call - ignore
+ return;
+ }
+
+ consumedSubpartitions[subpartitionIndex] = true;
+ remainingUnconsumed = (--numUnconsumedSubpartitions);
}
- LOG.debug("{}: Received release notification for subpartition
{}.",
- this, subpartitionIndex);
+ LOG.debug("{}: Received consumed notification for subpartition
{}.", this, subpartitionIndex);
+
+ if (remainingUnconsumed == 0) {
+ partitionManager.onConsumedPartition(this);
+ } else if (remainingUnconsumed < 0) {
+ throw new IllegalStateException("Received consume
notification even though all subpartitions are already consumed.");
+ }
}
@Override
public String toString() {
return "ReleaseOnConsumptionResultPartition " +
partitionId.toString() + " [" + partitionType + ", "
[flink] 02/03: [FLINK-13245][network] Make subpartition consumption notification independant
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit acd15d2527e7c8c137cb8719e15da3edc0a2c994
Author: Zhijiang
AuthorDate: Fri Jul 26 11:25:40 2019 +0200
[FLINK-13245][network] Make subpartition consumption notification
independant
---
.../ReleaseOnConsumptionResultPartition.java | 42 +++---
.../ReleaseOnConsumptionResultPartitionTest.java | 41 ++---
2 files changed, 67 insertions(+), 16 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
index 19ec681..766f500 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
@@ -22,7 +22,6 @@ import
org.apache.flink.runtime.io.network.buffer.BufferPoolOwner;
import org.apache.flink.util.function.FunctionWithException;
import java.io.IOException;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.flink.util.Preconditions.checkState;
@@ -31,11 +30,18 @@ import static
org.apache.flink.util.Preconditions.checkState;
*/
public class ReleaseOnConsumptionResultPartition extends ResultPartition {
+ private static final Object lock = new Object();
+
+ /**
+* A flag for each subpartition indicating whether it was already
consumed or not.
+*/
+ private final boolean[] consumedSubpartitions;
+
/**
* The total number of references to subpartitions of this result. The
result partition can be
* safely released, iff the reference count is zero.
*/
- private final AtomicInteger pendingReferences = new AtomicInteger();
+ private int numUnconsumedSubpartitions;
ReleaseOnConsumptionResultPartition(
String owningTaskName,
@@ -47,12 +53,13 @@ public class ReleaseOnConsumptionResultPartition extends
ResultPartition {
FunctionWithException bufferPoolFactory) {
super(owningTaskName, partitionId, partitionType,
subpartitions, numTargetKeyGroups, partitionManager, bufferPoolFactory);
- pendingReferences.set(subpartitions.length);
+ this.consumedSubpartitions = new boolean[subpartitions.length];
+ this.numUnconsumedSubpartitions = subpartitions.length;
}
@Override
public ResultSubpartitionView createSubpartitionView(int index,
BufferAvailabilityListener availabilityListener) throws IOException {
- checkState(pendingReferences.get() > 0, "Partition not
pinned.");
+ checkState(numUnconsumedSubpartitions > 0, "Partition not
pinned.");
return super.createSubpartitionView(index,
availabilityListener);
}
@@ -63,22 +70,33 @@ public class ReleaseOnConsumptionResultPartition extends
ResultPartition {
return;
}
- int refCnt = pendingReferences.decrementAndGet();
+ final int remainingUnconsumed;
- if (refCnt == 0) {
- partitionManager.onConsumedPartition(this);
- } else if (refCnt < 0) {
- throw new IllegalStateException("All references
released.");
+ // we synchronize only the bookkeeping section, to avoid
holding the lock during any
+ // calls into other components
+ synchronized (lock) {
+ if (consumedSubpartitions[subpartitionIndex]) {
+ // repeated call - ignore
+ return;
+ }
+
+ consumedSubpartitions[subpartitionIndex] = true;
+ remainingUnconsumed = (--numUnconsumedSubpartitions);
}
- LOG.debug("{}: Received release notification for subpartition
{}.",
- this, subpartitionIndex);
+ LOG.debug("{}: Received consumed notification for subpartition
{}.", this, subpartitionIndex);
+
+ if (remainingUnconsumed == 0) {
+ partitionManager.onConsumedPartition(this);
+ } else if (remainingUnconsumed < 0) {
+ throw new IllegalStateException("Received consume
notification even though all subpartitions are already consumed.");
+ }
}
@Override
public String toString() {
return "ReleaseOnConsumptionResultPartition " +
partitionId.toString() + " [" + partitionType + ", "
+ sub
[flink] 01/03: [FLINK-13245][network] Fix the bug of file resource leak while canceling partition request
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 653ba1a1bccb7457c7206c385a49d911b4950483
Author: Zhijiang
AuthorDate: Tue Jul 16 23:00:09 2019 +0800
[FLINK-13245][network] Fix the bug of file resource leak while canceling
partition request
On producer side the netty handler receives the CancelPartitionRequest for
releasing the SubpartitionView resource.
In previous implementation we try to find the corresponding view via
available queue in PartitionRequestQueue. But
in reality the view is not always available to stay in this queue, then the
view would never be released.
Furthermore the release of ResultPartition/ResultSubpartitions is based on
the reference counter in ReleaseOnConsumptionResultPartition,
but while handling the CancelPartitionRequest in PartitionRequestQueue, the
ReleaseOnConsumptionResultPartition is never
notified of consumed subpartition. That means the reference counter would
never decrease to 0 to trigger partition release,
which would bring file resource leak in the case of
BoundedBlockingSubpartition.
In order to fix above two issues, the corresponding view is released via
all reader queue instead, and then it would call
ReleaseOnConsumptionResultPartition#onConsumedSubpartition meanwhile to
solve this bug.
---
.../io/network/netty/PartitionRequestQueue.java| 33
.../network/netty/CancelPartitionRequestTest.java | 4 +-
.../network/netty/PartitionRequestQueueTest.java | 92 ++
.../io/network/partition/PartitionTestUtils.java | 8 ++
4 files changed, 118 insertions(+), 19 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index f82a42f..b492ea6 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -138,9 +138,7 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
}
for (NetworkSequenceViewReader reader : allReaders.values()) {
- reader.notifySubpartitionConsumed();
- reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
+ releaseViewReader(reader);
}
allReaders.clear();
}
@@ -181,19 +179,14 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
return;
}
- // Cancel the request for the input channel
- int size = availableReaders.size();
- for (int i = 0; i < size; i++) {
- NetworkSequenceViewReader reader =
pollAvailableReader();
- if (reader.getReceiverId().equals(toCancel)) {
- reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
- } else {
- registerAvailableReader(reader);
- }
- }
+ // remove reader from queue of available readers
+ availableReaders.removeIf(reader ->
reader.getReceiverId().equals(toCancel));
- allReaders.remove(toCancel);
+ // remove reader from queue of all readers and release
its resource
+ final NetworkSequenceViewReader toRelease =
allReaders.remove(toCancel);
+ if (toRelease != null) {
+ releaseViewReader(toRelease);
+ }
} else {
ctx.fireUserEventTriggered(msg);
}
@@ -308,14 +301,20 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
private void releaseAllResources() throws IOException {
// note: this is only ever executed by one thread: the Netty IO
thread!
for (NetworkSequenceViewReader reader : allReaders.values()) {
- reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
+ releaseViewReader(reader);
}
availableReaders.clear();
allReaders.clear();
}
+ private void releaseViewReader(NetworkSequenceViewReader reader) throws
IOException {
+ reader.notifySubpartitionConsumed();
+ reader.setRegisteredAsAvailable(false)
[flink] 03/03: [FLINK-13245][network] Remove redundant bookkeeping for already canceled input channel IDs
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 71a53d49d1c6ce0e5f840e1b528cb75323dc2665
Author: Zhijiang
AuthorDate: Fri Jul 26 11:50:55 2019 +0200
[FLINK-13245][network] Remove redundant bookkeeping for already canceled
input channel IDs
---
.../runtime/io/network/netty/PartitionRequestQueue.java | 16
1 file changed, 16 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index b492ea6..4a845d1 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -26,7 +26,6 @@ import
org.apache.flink.runtime.io.network.partition.ProducerFailedException;
import
org.apache.flink.runtime.io.network.partition.consumer.InputChannel.BufferAndAvailability;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
-import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
@@ -40,7 +39,6 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayDeque;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -62,8 +60,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
/** All the readers created for the consumers' partition requests. */
private final ConcurrentMap
allReaders = new ConcurrentHashMap<>();
- private final Set released = Sets.newHashSet();
-
private boolean fatalError;
private ChannelHandlerContext ctx;
@@ -175,9 +171,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
} else if (msg.getClass() == InputChannelID.class) {
// Release partition view that get a cancel request.
InputChannelID toCancel = (InputChannelID) msg;
- if (released.contains(toCancel)) {
- return;
- }
// remove reader from queue of available readers
availableReaders.removeIf(reader ->
reader.getReceiverId().equals(toCancel));
@@ -222,7 +215,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
if (!reader.isReleased()) {
continue;
}
- markAsReleased(reader.getReceiverId());
Throwable cause =
reader.getFailureCause();
if (cause != null) {
@@ -312,14 +304,6 @@ class PartitionRequestQueue extends
ChannelInboundHandlerAdapter {
reader.notifySubpartitionConsumed();
reader.setRegisteredAsAvailable(false);
reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
- }
-
- /**
-* Marks a receiver as released.
-*/
- private void markAsReleased(InputChannelID receiverId) {
- released.add(receiverId);
}
// This listener is called after an element of the current
nonEmptyReader has been
[flink] branch master updated (12118b9 -> 71a53d4)
This is an automated email from the ASF dual-hosted git repository. sewen pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 12118b9 [FLINK-13387][WebUI] Fix log download for old UI new 653ba1a [FLINK-13245][network] Fix the bug of file resource leak while canceling partition request new acd15d2 [FLINK-13245][network] Make subpartition consumption notification independant new 71a53d4 [FLINK-13245][network] Remove redundant bookkeeping for already canceled input channel IDs The 3 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../io/network/netty/PartitionRequestQueue.java| 43 +++--- .../ReleaseOnConsumptionResultPartition.java | 42 +++--- .../network/netty/CancelPartitionRequestTest.java | 4 +- .../network/netty/PartitionRequestQueueTest.java | 92 ++ .../io/network/partition/PartitionTestUtils.java | 8 ++ .../ReleaseOnConsumptionResultPartitionTest.java | 41 +- 6 files changed, 182 insertions(+), 48 deletions(-)
[flink] branch master updated: [hotfix] Fix typo in api_concepts.md
This is an automated email from the ASF dual-hosted git repository. aljoscha pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/master by this push: new db6529a [hotfix] Fix typo in api_concepts.md db6529a is described below commit db6529aebc74b0a0d3422ecc0fc86cb1c135707d Author: WangHengwei <35023...@qq.com> AuthorDate: Mon Jul 29 19:43:34 2019 +0800 [hotfix] Fix typo in api_concepts.md --- docs/dev/api_concepts.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/dev/api_concepts.md b/docs/dev/api_concepts.md index b70ab08..1e8fd18 100644 --- a/docs/dev/api_concepts.md +++ b/docs/dev/api_concepts.md @@ -758,7 +758,7 @@ Restrictions apply to classes containing fields that cannot be serialized, like resources. Classes that follow the Java Beans conventions work well in general. All classes that are not identified as POJO types (see POJO requirements above) are handled by Flink as general class types. -Flink treats these data types as black boxes and is not able to access their content (i.e., for efficient sorting). General types are de/serialized using the serialization framework [Kryo](https://github.com/EsotericSoftware/kryo). +Flink treats these data types as black boxes and is not able to access their content (e.g., for efficient sorting). General types are de/serialized using the serialization framework [Kryo](https://github.com/EsotericSoftware/kryo). Values
[flink] branch release-1.9 updated: [hotfix] Fix typo in api_concepts.md
This is an automated email from the ASF dual-hosted git repository. aljoscha pushed a commit to branch release-1.9 in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/release-1.9 by this push: new 41675a9 [hotfix] Fix typo in api_concepts.md 41675a9 is described below commit 41675a9ffda09056d9d2490681475dd044532d40 Author: WangHengwei <35023...@qq.com> AuthorDate: Mon Jul 29 19:43:34 2019 +0800 [hotfix] Fix typo in api_concepts.md --- docs/dev/api_concepts.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/dev/api_concepts.md b/docs/dev/api_concepts.md index b70ab08..1e8fd18 100644 --- a/docs/dev/api_concepts.md +++ b/docs/dev/api_concepts.md @@ -758,7 +758,7 @@ Restrictions apply to classes containing fields that cannot be serialized, like resources. Classes that follow the Java Beans conventions work well in general. All classes that are not identified as POJO types (see POJO requirements above) are handled by Flink as general class types. -Flink treats these data types as black boxes and is not able to access their content (i.e., for efficient sorting). General types are de/serialized using the serialization framework [Kryo](https://github.com/EsotericSoftware/kryo). +Flink treats these data types as black boxes and is not able to access their content (e.g., for efficient sorting). General types are de/serialized using the serialization framework [Kryo](https://github.com/EsotericSoftware/kryo). Values
[flink] branch master updated: [FLINK-13458][table] ThreadLocalCache clashes for Blink planner
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 14afeea [FLINK-13458][table] ThreadLocalCache clashes for Blink
planner
14afeea is described below
commit 14afeea06dcd9cb90dfaf471156b34a14380038b
Author: Timo Walther
AuthorDate: Mon Jul 29 09:32:48 2019 +0200
[FLINK-13458][table] ThreadLocalCache clashes for Blink planner
This closes #9257.
---
.../flink/table/utils}/ThreadLocalCache.java | 10 +++--
.../runtime/functions/DateTimeFunctions.scala | 1 +
.../table/runtime/functions/ThreadLocalCache.scala | 49 --
.../table/runtime/functions/SqlDateTimeUtils.java | 1 +
.../table/runtime/functions/SqlFunctionUtils.java | 1 +
5 files changed, 10 insertions(+), 52 deletions(-)
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/ThreadLocalCache.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/ThreadLocalCache.java
similarity index 90%
rename from
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/ThreadLocalCache.java
rename to
flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/ThreadLocalCache.java
index 39de4df..ca47ab0 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/ThreadLocalCache.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/ThreadLocalCache.java
@@ -16,15 +16,19 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.functions;
+package org.apache.flink.table.utils;
+
+import org.apache.flink.annotation.Internal;
import java.util.LinkedHashMap;
import java.util.Map;
/**
- * Provides a ThreadLocal cache with a maximum cache size per thread.
- * Values must not be null.
+ * Provides a thread local cache with a maximum cache size per thread.
+ *
+ * Note: Values must not be null.
*/
+@Internal
public abstract class ThreadLocalCache {
private static final int DEFAULT_CACHE_SIZE = 64;
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
index d69a5c9..0bde983 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
@@ -17,6 +17,7 @@
*/
package org.apache.flink.table.runtime.functions
+import org.apache.flink.table.utils.ThreadLocalCache
import org.joda.time.format.DateTimeFormatter
import org.joda.time.format.DateTimeFormatterBuilder
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ThreadLocalCache.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ThreadLocalCache.scala
deleted file mode 100644
index b3a8d7a..000
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ThreadLocalCache.scala
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.flink.table.runtime.functions
-
-import java.util.{LinkedHashMap => JLinkedHashMap}
-import java.util.{Map => JMap}
-
-/**
- * Provides a ThreadLocal cache with a maximum cache size per thread.
- * Values must not be null.
- */
-abstract class ThreadLocalCache[K, V](val maxSizePerThread: Int) {
- private val cache = new ThreadLocal[BoundedMap[K, V]]
-
- protected def getNewInstance(key: K): V
-
- def get(key: K): V = {
-var m = cache.get
-if (m == null) {
- m = new BoundedMap(maxSizePerThread)
- cache.set(m)
-}
-var v = m.get(key)
-if (v == null) {
- v = getNewInstance(key)
- m.put(key, v)
-}
-v
- }
-}
-
-private class BoundedMap[K, V](val
[flink] branch release-1.9 updated: [FLINK-13458][table] ThreadLocalCache clashes for Blink planner
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 36a7fde [FLINK-13458][table] ThreadLocalCache clashes for Blink
planner
36a7fde is described below
commit 36a7fde092df89e3be7660075df3dcdfafc762b6
Author: Timo Walther
AuthorDate: Mon Jul 29 09:32:48 2019 +0200
[FLINK-13458][table] ThreadLocalCache clashes for Blink planner
This closes #9257.
---
.../flink/table/utils}/ThreadLocalCache.java | 10 +++--
.../runtime/functions/DateTimeFunctions.scala | 1 +
.../table/runtime/functions/ThreadLocalCache.scala | 49 --
.../table/runtime/functions/SqlDateTimeUtils.java | 1 +
.../table/runtime/functions/SqlFunctionUtils.java | 1 +
5 files changed, 10 insertions(+), 52 deletions(-)
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/ThreadLocalCache.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/ThreadLocalCache.java
similarity index 90%
rename from
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/ThreadLocalCache.java
rename to
flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/ThreadLocalCache.java
index 39de4df..ca47ab0 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/ThreadLocalCache.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/ThreadLocalCache.java
@@ -16,15 +16,19 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.functions;
+package org.apache.flink.table.utils;
+
+import org.apache.flink.annotation.Internal;
import java.util.LinkedHashMap;
import java.util.Map;
/**
- * Provides a ThreadLocal cache with a maximum cache size per thread.
- * Values must not be null.
+ * Provides a thread local cache with a maximum cache size per thread.
+ *
+ * Note: Values must not be null.
*/
+@Internal
public abstract class ThreadLocalCache {
private static final int DEFAULT_CACHE_SIZE = 64;
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
index d69a5c9..0bde983 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/DateTimeFunctions.scala
@@ -17,6 +17,7 @@
*/
package org.apache.flink.table.runtime.functions
+import org.apache.flink.table.utils.ThreadLocalCache
import org.joda.time.format.DateTimeFormatter
import org.joda.time.format.DateTimeFormatterBuilder
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ThreadLocalCache.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ThreadLocalCache.scala
deleted file mode 100644
index b3a8d7a..000
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ThreadLocalCache.scala
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.flink.table.runtime.functions
-
-import java.util.{LinkedHashMap => JLinkedHashMap}
-import java.util.{Map => JMap}
-
-/**
- * Provides a ThreadLocal cache with a maximum cache size per thread.
- * Values must not be null.
- */
-abstract class ThreadLocalCache[K, V](val maxSizePerThread: Int) {
- private val cache = new ThreadLocal[BoundedMap[K, V]]
-
- protected def getNewInstance(key: K): V
-
- def get(key: K): V = {
-var m = cache.get
-if (m == null) {
- m = new BoundedMap(maxSizePerThread)
- cache.set(m)
-}
-var v = m.get(key)
-if (v == null) {
- v = getNewInstance(key)
- m.put(key, v)
-}
-v
- }
-}
-
-private class BoundedMap[
[flink] branch master updated: [FLINK-13228][tests][filesystems] Harden HadoopRecoverableWriterTest
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e373c44 [FLINK-13228][tests][filesystems] Harden
HadoopRecoverableWriterTest
e373c44 is described below
commit e373c4481e6a0ca0e1e73a6170b9e3da5cc9be5b
Author: Yu Li
AuthorDate: Fri Jul 26 10:27:28 2019 +0200
[FLINK-13228][tests][filesystems] Harden HadoopRecoverableWriterTest
Currently test cases will fail when trying to close the output stream if
all data written
but ClosedByInterruptException occurs at the ending phase. This commit
fixes it.
This closes #9235
---
.../core/fs/AbstractRecoverableWriterTest.java | 49 ++
1 file changed, 41 insertions(+), 8 deletions(-)
diff --git
a/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
b/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
index ab37a07..de9b095 100644
---
a/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
+++
b/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.core.fs;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.IOUtils;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TestLogger;
@@ -111,7 +112,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final Path path = new Path(testDir, "part-0");
- try (final RecoverableFsDataOutputStream stream =
writer.open(path)) {
+ RecoverableFsDataOutputStream stream = null;
+ try {
+ stream = writer.open(path);
stream.write(testData1.getBytes(StandardCharsets.UTF_8));
stream.closeForCommit().commit();
@@ -119,6 +122,8 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
Assert.assertEquals("part-0",
fileContents.getKey().getName());
Assert.assertEquals(testData1,
fileContents.getValue());
}
+ } finally {
+ IOUtils.closeQuietly(stream);
}
}
@@ -130,7 +135,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final Path path = new Path(testDir, "part-0");
- try (final RecoverableFsDataOutputStream stream =
writer.open(path)) {
+ RecoverableFsDataOutputStream stream = null;
+ try {
+ stream = writer.open(path);
stream.write(testData1.getBytes(StandardCharsets.UTF_8));
stream.persist();
@@ -141,6 +148,8 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
Assert.assertEquals("part-0",
fileContents.getKey().getName());
Assert.assertEquals(testData1 + testData2,
fileContents.getValue());
}
+ } finally {
+ IOUtils.closeQuietly(stream);
}
}
@@ -194,7 +203,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final RecoverableWriter initWriter = getNewFileSystemWriter();
final Map
recoverables = new HashMap<>(4);
- try (final RecoverableFsDataOutputStream stream =
initWriter.open(path)) {
+ RecoverableFsDataOutputStream stream = null;
+ try {
+ stream = initWriter.open(path);
recoverables.put(INIT_EMPTY_PERSIST, stream.persist());
stream.write(testData1.getBytes(StandardCharsets.UTF_8));
@@ -206,6 +217,8 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
stream.write(testData2.getBytes(StandardCharsets.UTF_8));
recoverables.put(FINAL_WITH_EXTRA_STATE,
stream.persist());
+ } finally {
+ IOUtils.closeQuietly(stream);
}
final
SimpleVersionedSerializer serializer =
initWriter.getResumeRecoverableSerializer();
@@ -217,7 +230,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final RecoverableWriter.ResumeRecoverable recoveredRecoverable =
deserializer.deserialize(serializer.getVersion(), serializedRecoverable);
- try (final RecoverableFsDataOutputStream recoveredStream =
newWriter.recover(recoveredRecoverable)) {
+ RecoverableFsDataOutputStream recoveredStream = null;
+
[flink] branch release-1.9 updated: [FLINK-13228][tests][filesystems] Harden HadoopRecoverableWriterTest
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 0e9f463 [FLINK-13228][tests][filesystems] Harden
HadoopRecoverableWriterTest
0e9f463 is described below
commit 0e9f463668378bd7469194ebf0af76e3c125f0d7
Author: Yu Li
AuthorDate: Fri Jul 26 10:27:28 2019 +0200
[FLINK-13228][tests][filesystems] Harden HadoopRecoverableWriterTest
Currently test cases will fail when trying to close the output stream if
all data written
but ClosedByInterruptException occurs at the ending phase. This commit
fixes it.
This closes #9235
---
.../core/fs/AbstractRecoverableWriterTest.java | 49 ++
1 file changed, 41 insertions(+), 8 deletions(-)
diff --git
a/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
b/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
index ab37a07..de9b095 100644
---
a/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
+++
b/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.core.fs;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.IOUtils;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TestLogger;
@@ -111,7 +112,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final Path path = new Path(testDir, "part-0");
- try (final RecoverableFsDataOutputStream stream =
writer.open(path)) {
+ RecoverableFsDataOutputStream stream = null;
+ try {
+ stream = writer.open(path);
stream.write(testData1.getBytes(StandardCharsets.UTF_8));
stream.closeForCommit().commit();
@@ -119,6 +122,8 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
Assert.assertEquals("part-0",
fileContents.getKey().getName());
Assert.assertEquals(testData1,
fileContents.getValue());
}
+ } finally {
+ IOUtils.closeQuietly(stream);
}
}
@@ -130,7 +135,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final Path path = new Path(testDir, "part-0");
- try (final RecoverableFsDataOutputStream stream =
writer.open(path)) {
+ RecoverableFsDataOutputStream stream = null;
+ try {
+ stream = writer.open(path);
stream.write(testData1.getBytes(StandardCharsets.UTF_8));
stream.persist();
@@ -141,6 +148,8 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
Assert.assertEquals("part-0",
fileContents.getKey().getName());
Assert.assertEquals(testData1 + testData2,
fileContents.getValue());
}
+ } finally {
+ IOUtils.closeQuietly(stream);
}
}
@@ -194,7 +203,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final RecoverableWriter initWriter = getNewFileSystemWriter();
final Map
recoverables = new HashMap<>(4);
- try (final RecoverableFsDataOutputStream stream =
initWriter.open(path)) {
+ RecoverableFsDataOutputStream stream = null;
+ try {
+ stream = initWriter.open(path);
recoverables.put(INIT_EMPTY_PERSIST, stream.persist());
stream.write(testData1.getBytes(StandardCharsets.UTF_8));
@@ -206,6 +217,8 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
stream.write(testData2.getBytes(StandardCharsets.UTF_8));
recoverables.put(FINAL_WITH_EXTRA_STATE,
stream.persist());
+ } finally {
+ IOUtils.closeQuietly(stream);
}
final
SimpleVersionedSerializer serializer =
initWriter.getResumeRecoverableSerializer();
@@ -217,7 +230,9 @@ public abstract class AbstractRecoverableWriterTest extends
TestLogger {
final RecoverableWriter.ResumeRecoverable recoveredRecoverable =
deserializer.deserialize(serializer.getVersion(), serializedRecoverable);
- try (final RecoverableFsDataOutputStream recoveredStream =
newWriter.recover(recoveredRecoverable)) {
+ RecoverableFsDataOutputStream recoveredStream = nu
[flink] branch master updated: [FLINK-13447][table] Change default planner to legacy planner instead of any one
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5b01e71 [FLINK-13447][table] Change default planner to legacy planner
instead of any one
5b01e71 is described below
commit 5b01e7139c1d0307e93f0e448e75ee6b5e6541ca
Author: Jark Wu
AuthorDate: Sat Jul 27 23:08:18 2019 +0800
[FLINK-13447][table] Change default planner to legacy planner instead of
any one
This closes #9249
---
.../flink/table/api/EnvironmentSettings.java | 26 +---
.../flink/table/api/TableEnvironmentTest.scala | 2 +-
.../planner/match/PatternTranslatorTestBase.scala | 3 +-
.../validation/SetOperatorsValidationTest.scala| 11 ---
.../table/validation/TableSinkValidationTest.scala | 9 +++---
.../validation/UnsupportedOpsValidationTest.scala | 20 ++---
.../planner/plan/utils/FlinkRelOptUtilTest.scala | 3 +-
.../runtime/stream/sql/MatchRecognizeITCase.scala | 35 +++---
.../runtime/stream/sql/TemporalJoinITCase.scala| 6 ++--
.../runtime/stream/table/TableSinkITCase.scala | 29 +-
.../runtime/utils/StreamingWithStateTestBase.scala | 4 +--
.../flink/table/planner/utils/TableTestBase.scala | 5
12 files changed, 79 insertions(+), 74 deletions(-)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
index 29c9227..8289bbc 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
@@ -149,29 +149,35 @@ public class EnvironmentSettings {
* A builder for {@link EnvironmentSettings}.
*/
public static class Builder {
- private String plannerClass = null;
- private String executorClass = null;
+ private static final String OLD_PLANNER_FACTORY =
"org.apache.flink.table.planner.StreamPlannerFactory";
+ private static final String OLD_EXECUTOR_FACTORY =
"org.apache.flink.table.executor.StreamExecutorFactory";
+ private static final String BLINK_PLANNER_FACTORY =
"org.apache.flink.table.planner.delegation.BlinkPlannerFactory";
+ private static final String BLINK_EXECUTOR_FACTORY =
"org.apache.flink.table.planner.delegation.BlinkExecutorFactory";
+
+ private String plannerClass = OLD_PLANNER_FACTORY;
+ private String executorClass = OLD_EXECUTOR_FACTORY;
private String builtInCatalogName = "default_catalog";
private String builtInDatabaseName = "default_database";
private boolean isStreamingMode = true;
/**
-* Sets the old Flink planner as the required module. By
default, {@link #useAnyPlanner()} is
-* enabled.
+* Sets the old Flink planner as the required module.
+*
+* This is the default behavior.
*/
public Builder useOldPlanner() {
- this.plannerClass =
"org.apache.flink.table.planner.StreamPlannerFactory";
- this.executorClass =
"org.apache.flink.table.executor.StreamExecutorFactory";
+ this.plannerClass = OLD_PLANNER_FACTORY;
+ this.executorClass = OLD_EXECUTOR_FACTORY;
return this;
}
/**
-* Sets the Blink planner as the required module. By default,
{@link #useAnyPlanner()} is
+* Sets the Blink planner as the required module. By default,
{@link #useOldPlanner()} is
* enabled.
*/
public Builder useBlinkPlanner() {
- this.plannerClass =
"org.apache.flink.table.planner.delegation.BlinkPlannerFactory";
- this.executorClass =
"org.apache.flink.table.planner.delegation.BlinkExecutorFactory";
+ this.plannerClass = BLINK_PLANNER_FACTORY;
+ this.executorClass = BLINK_EXECUTOR_FACTORY;
return this;
}
@@ -180,7 +186,7 @@ public class EnvironmentSettings {
*
* A planner will be discovered automatically, if there is
only one planner available.
*
-* This is the default behavior.
+* By default, {@link #useOldPlanner()} is enabled.
*/
public Builder useAnyPlanner() {
this.plannerClass = null;
diff --git
a/flink-table/flink-table
[flink] branch release-1.9 updated: [FLINK-13447][table] Change default planner to legacy planner instead of any one
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new c763cef [FLINK-13447][table] Change default planner to legacy planner
instead of any one
c763cef is described below
commit c763cef2f0d244b305fc9e3c70be0d6c89d5eba5
Author: Jark Wu
AuthorDate: Sat Jul 27 23:08:18 2019 +0800
[FLINK-13447][table] Change default planner to legacy planner instead of
any one
This closes #9249
---
.../flink/table/api/EnvironmentSettings.java | 26 +---
.../flink/table/api/TableEnvironmentTest.scala | 2 +-
.../planner/match/PatternTranslatorTestBase.scala | 3 +-
.../validation/SetOperatorsValidationTest.scala| 11 ---
.../table/validation/TableSinkValidationTest.scala | 9 +++---
.../validation/UnsupportedOpsValidationTest.scala | 20 ++---
.../planner/plan/utils/FlinkRelOptUtilTest.scala | 3 +-
.../runtime/stream/sql/MatchRecognizeITCase.scala | 35 +++---
.../runtime/stream/sql/TemporalJoinITCase.scala| 6 ++--
.../runtime/stream/table/TableSinkITCase.scala | 29 +-
.../runtime/utils/StreamingWithStateTestBase.scala | 4 +--
.../flink/table/planner/utils/TableTestBase.scala | 5
12 files changed, 79 insertions(+), 74 deletions(-)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
index 29c9227..8289bbc 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
@@ -149,29 +149,35 @@ public class EnvironmentSettings {
* A builder for {@link EnvironmentSettings}.
*/
public static class Builder {
- private String plannerClass = null;
- private String executorClass = null;
+ private static final String OLD_PLANNER_FACTORY =
"org.apache.flink.table.planner.StreamPlannerFactory";
+ private static final String OLD_EXECUTOR_FACTORY =
"org.apache.flink.table.executor.StreamExecutorFactory";
+ private static final String BLINK_PLANNER_FACTORY =
"org.apache.flink.table.planner.delegation.BlinkPlannerFactory";
+ private static final String BLINK_EXECUTOR_FACTORY =
"org.apache.flink.table.planner.delegation.BlinkExecutorFactory";
+
+ private String plannerClass = OLD_PLANNER_FACTORY;
+ private String executorClass = OLD_EXECUTOR_FACTORY;
private String builtInCatalogName = "default_catalog";
private String builtInDatabaseName = "default_database";
private boolean isStreamingMode = true;
/**
-* Sets the old Flink planner as the required module. By
default, {@link #useAnyPlanner()} is
-* enabled.
+* Sets the old Flink planner as the required module.
+*
+* This is the default behavior.
*/
public Builder useOldPlanner() {
- this.plannerClass =
"org.apache.flink.table.planner.StreamPlannerFactory";
- this.executorClass =
"org.apache.flink.table.executor.StreamExecutorFactory";
+ this.plannerClass = OLD_PLANNER_FACTORY;
+ this.executorClass = OLD_EXECUTOR_FACTORY;
return this;
}
/**
-* Sets the Blink planner as the required module. By default,
{@link #useAnyPlanner()} is
+* Sets the Blink planner as the required module. By default,
{@link #useOldPlanner()} is
* enabled.
*/
public Builder useBlinkPlanner() {
- this.plannerClass =
"org.apache.flink.table.planner.delegation.BlinkPlannerFactory";
- this.executorClass =
"org.apache.flink.table.planner.delegation.BlinkExecutorFactory";
+ this.plannerClass = BLINK_PLANNER_FACTORY;
+ this.executorClass = BLINK_EXECUTOR_FACTORY;
return this;
}
@@ -180,7 +186,7 @@ public class EnvironmentSettings {
*
* A planner will be discovered automatically, if there is
only one planner available.
*
-* This is the default behavior.
+* By default, {@link #useOldPlanner()} is enabled.
*/
public Builder useAnyPlanner() {
this.plannerClass = null;
diff --git
a/flink-table/f
[flink] branch master updated: [FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7fec439 [FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
7fec439 is described below
commit 7fec4392b21b07c69ba15ea554731886f181609e
Author: ifndef-SleePy
AuthorDate: Tue Jul 23 14:29:36 2019 +0200
[FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
---
flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
index a22c6d4..2ecea72 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
@@ -21,8 +21,9 @@ source "$(dirname "$0")"/common.sh
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-bucketing-sink-test/target/BucketingSinkTestProgram.jar
-# enable DEBUG logging level to retrieve truncate length later
-sed -i -e
's/#log4j.logger.org.apache.flink=INFO/log4j.logger.org.apache.flink=DEBUG/g'
$FLINK_DIR/conf/log4j.properties
+# enable DEBUG logging level for the BucketingSink to retrieve truncate length
later
+echo "" >> $FLINK_DIR/conf/log4j.properties
+echo
"log4j.logger.org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink=DEBUG"
>> $FLINK_DIR/conf/log4j.properties
set_conf_ssl
start_cluster
[flink] branch release-1.9 updated: [FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 32ec971 [FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
32ec971 is described below
commit 32ec971b3617e8fe4fc87334f747ed44afceb4c2
Author: ifndef-SleePy
AuthorDate: Tue Jul 23 14:29:36 2019 +0200
[FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
---
flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
index a22c6d4..2ecea72 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
@@ -21,8 +21,9 @@ source "$(dirname "$0")"/common.sh
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-bucketing-sink-test/target/BucketingSinkTestProgram.jar
-# enable DEBUG logging level to retrieve truncate length later
-sed -i -e
's/#log4j.logger.org.apache.flink=INFO/log4j.logger.org.apache.flink=DEBUG/g'
$FLINK_DIR/conf/log4j.properties
+# enable DEBUG logging level for the BucketingSink to retrieve truncate length
later
+echo "" >> $FLINK_DIR/conf/log4j.properties
+echo
"log4j.logger.org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink=DEBUG"
>> $FLINK_DIR/conf/log4j.properties
set_conf_ssl
start_cluster
[flink] branch release-1.8 updated: [FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch release-1.8
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.8 by this push:
new c4e9743 [FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
c4e9743 is described below
commit c4e97436bb0ef5a658ff4a9117b4776d58f888c0
Author: ifndef-SleePy
AuthorDate: Tue Jul 23 14:29:36 2019 +0200
[FLINK-9526][e2e] Fix unstable BucketingSink end-to-end test
---
flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
index ba43e21..b3d19fd 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
@@ -21,8 +21,9 @@ source "$(dirname "$0")"/common.sh
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-bucketing-sink-test/target/BucketingSinkTestProgram.jar
-# enable DEBUG logging level to retrieve truncate length later
-sed -i -e
's/#log4j.logger.org.apache.flink=INFO/log4j.logger.org.apache.flink=DEBUG/g'
$FLINK_DIR/conf/log4j.properties
+# enable DEBUG logging level for the BucketingSink to retrieve truncate length
later
+echo "" >> $FLINK_DIR/conf/log4j.properties
+echo
"log4j.logger.org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink=DEBUG"
>> $FLINK_DIR/conf/log4j.properties
set_conf_ssl
start_cluster
[flink-web] branch asf-site updated: add myself to the flink community page
This is an automated email from the ASF dual-hosted git repository. rongr pushed a commit to branch asf-site in repository https://gitbox.apache.org/repos/asf/flink-web.git The following commit(s) were added to refs/heads/asf-site by this push: new 3acaa86 add myself to the flink community page 3acaa86 is described below commit 3acaa862cb998345da10ffb8bfff754d77e16d6a Author: Rong Rong AuthorDate: Sun Jul 28 16:48:46 2019 -0700 add myself to the flink community page --- community.md | 6 ++ community.zh.md | 6 ++ content/community.html| 6 ++ content/zh/community.html | 6 ++ 4 files changed, 24 insertions(+) diff --git a/community.md b/community.md index 4fd0b12..535f9ba 100644 --- a/community.md +++ b/community.md @@ -329,6 +329,12 @@ Flink Forward is a conference happening yearly in different locations around the trohrmann +https://avatars1.githubusercontent.com/u/3581352?s=50"; class="committer-avatar"> +Rong Rong +Committer +rongr + + https://avatars0.githubusercontent.com/u/105434?s=50"; class="committer-avatar"> Henry Saputra PMC, Committer diff --git a/community.zh.md b/community.zh.md index b596f1c..6fd0be6 100644 --- a/community.zh.md +++ b/community.zh.md @@ -324,6 +324,12 @@ Flink Forward 大会每年都会在世界的不同地方举办。关于大会最 trohrmann +https://avatars1.githubusercontent.com/u/3581352?s=50"; class="committer-avatar"> +Rong Rong +Committer +rongr + + https://avatars0.githubusercontent.com/u/105434?s=50"; class="committer-avatar"> Henry Saputra PMC, Committer diff --git a/content/community.html b/content/community.html index 08d811a..8aa79b3 100644 --- a/content/community.html +++ b/content/community.html @@ -517,6 +517,12 @@ trohrmann +https://avatars1.githubusercontent.com/u/3581352?s=50"; class="committer-avatar" /> +Rong Rong +Committer +rongr + + https://avatars0.githubusercontent.com/u/105434?s=50"; class="committer-avatar" /> Henry Saputra PMC, Committer diff --git a/content/zh/community.html b/content/zh/community.html index a760b42..6688913 100644 --- a/content/zh/community.html +++ b/content/zh/community.html @@ -514,6 +514,12 @@ trohrmann +https://avatars1.githubusercontent.com/u/3581352?s=50"; class="committer-avatar" /> +Rong Rong +Committer +rongr + + https://avatars0.githubusercontent.com/u/105434?s=50"; class="committer-avatar" /> Henry Saputra PMC, Committer
[flink] branch master updated: [FLINK-12747][docs] Getting Started - Table Api Walkthrough
This is an automated email from the ASF dual-hosted git repository.
nkruber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f4943dd [FLINK-12747][docs] Getting Started - Table Api Walkthrough
f4943dd is described below
commit f4943dd06b2ff7da899812a2aaa2e0b24c2afc01
Author: Seth Wiesman
AuthorDate: Wed Jun 26 14:48:38 2019 -0500
[FLINK-12747][docs] Getting Started - Table Api Walkthrough
---
docs/getting-started/examples/index.md | 2 +-
docs/getting-started/examples/index.zh.md | 2 +-
docs/getting-started/tutorials/index.md| 2 +-
docs/getting-started/tutorials/index.zh.md | 2 +-
.../{tutorials => walkthroughs}/index.md | 8 +-
.../{tutorials => walkthroughs}/index.zh.md| 8 +-
docs/getting-started/walkthroughs/table_api.md | 494 +
docs/getting-started/walkthroughs/table_api.zh.md | 494 +
flink-end-to-end-tests/run-nightly-tests.sh| 3 +
.../test-scripts/test_table_walkthroughs.sh| 74 +++
.../flink-walkthrough-common/pom.xml | 59 +++
.../walkthrough/common/entity/Transaction.java | 93
.../common/sink/LoggerOutputFormat.java| 50 +++
.../common/source/TransactionIterator.java | 130 ++
.../common/source/TransactionRowInputFormat.java | 60 +++
.../common/source/TransactionSource.java | 65 +++
.../table/BoundedTransactionTableSource.java | 55 +++
.../common/table/SpendReportTableSink.java | 95
.../common/table/TruncateDateToHour.java | 46 ++
.../table/UnboundedTransactionTableSource.java | 89
.../flink-walkthrough-table-java/pom.xml | 26 +-
.../META-INF/maven/archetype-metadata.xml | 25 +-
.../src/main/resources/archetype-resources/pom.xml | 263 +++
.../src/main/java/SpendReport.java | 45 ++
.../src/main/resources/log4j.properties| 24 +
.../flink-walkthrough-table-scala/pom.xml | 26 +-
.../META-INF/maven/archetype-metadata.xml | 25 +-
.../src/main/resources/archetype-resources/pom.xml | 300 +
.../src/main/resources/log4j.properties| 24 +
.../src/main/scala/SpendReport.scala | 41 ++
flink-walkthroughs/pom.xml | 95
pom.xml| 1 +
32 files changed, 2686 insertions(+), 40 deletions(-)
diff --git a/docs/getting-started/examples/index.md
b/docs/getting-started/examples/index.md
index 6810bbd..d4d315c 100644
--- a/docs/getting-started/examples/index.md
+++ b/docs/getting-started/examples/index.md
@@ -3,7 +3,7 @@ title: Examples
nav-id: examples
nav-title: ' Examples'
nav-parent_id: getting-started
-nav-pos: 2
+nav-pos: 3
nav-show_overview: true
---
+-->
\ No newline at end of file
diff --git a/docs/getting-started/tutorials/index.zh.md
b/docs/getting-started/walkthroughs/index.zh.md
similarity index 84%
copy from docs/getting-started/tutorials/index.zh.md
copy to docs/getting-started/walkthroughs/index.zh.md
index 540a6c4..ae2f536 100644
--- a/docs/getting-started/tutorials/index.zh.md
+++ b/docs/getting-started/walkthroughs/index.zh.md
@@ -1,7 +1,7 @@
---
-title: "教程"
-nav-id: tutorials
-nav-title: '
教程'
+title: "Code Walkthroughs"
+nav-id: walkthroughs
+nav-title: ' Code
Walkthroughs'
nav-parent_id: getting-started
nav-pos: 1
---
@@ -22,4 +22,4 @@ software distributed under the License is distributed on an
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
--->
+-->
\ No newline at end of file
diff --git a/docs/getting-started/walkthroughs/table_api.md
b/docs/getting-started/walkthroughs/table_api.md
new file mode 100644
index 000..878fb54
--- /dev/null
+++ b/docs/getting-started/walkthroughs/table_api.md
@@ -0,0 +1,494 @@
+---
+title: "Table API"
+nav-id: tableapiwalkthrough
+nav-title: 'Table API'
+nav-parent_id: walkthroughs
+nav-pos: 1
+---
+
+
+Apache Flink offers a Table API as a unified, relational API for batch and
stream processing, i.e., queries are executed with the same semantics on
unbounded, real-time streams or bounded, batch data sets and produce the same
results.
+The Table API in Flink is commonly used to ease the definition of data
analytics, data pipelining, and ETL applications.
+
+* This will be replaced by the TOC
+{:toc}
+
+## What Will You Be Building?
+
+In this tutorial, you will learn how to build a continuous ETL pipeline for
tracking financial transactions by account over time.
+You will start by building your report as a nightly batch job, and then
migrate to a streaming pipeline.
+
+## Prerequisites
+
+This walkthrough assumes that you have some familiarity with Java or Scala,
but you s
buildbot failure in on flink-docs-master
The Buildbot has detected a new failure on builder flink-docs-master while building . Full details are available at: https://ci.apache.org/builders/flink-docs-master/builds/1549 Buildbot URL: https://ci.apache.org/ Buildslave for this Build: bb_slave2_ubuntu Build Reason: The Nightly scheduler named 'flink-nightly-docs-master' triggered this build Build Source Stamp: [branch master] HEAD Blamelist: BUILD FAILED: failed Build Python docs Sincerely, -The Buildbot
[flink] branch master updated: [FLINK-13347][table-planner] should handle SEMI/ANTI JoinRelType in switch case
This is an automated email from the ASF dual-hosted git repository.
rongr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ef29f30 [FLINK-13347][table-planner] should handle SEMI/ANTI
JoinRelType in switch case
ef29f30 is described below
commit ef29f305cd3d907d7c445c271b314ea643baaeeb
Author: godfreyhe
AuthorDate: Thu Jul 25 17:06:50 2019 +0800
[FLINK-13347][table-planner] should handle SEMI/ANTI JoinRelType in switch
case
This closes #9227.
---
.../scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala | 2 +-
.../main/scala/org/apache/flink/table/plan/nodes/CommonJoin.scala | 2 ++
.../org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala | 1 +
.../nodes/datastream/DataStreamJoinToCoProcessTranslator.scala| 8 +---
.../flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala | 2 ++
.../org/apache/flink/table/runtime/join/WindowJoinUtil.scala | 4 ++--
6 files changed, 13 insertions(+), 6 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
index bd2e8fa..923307c 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
@@ -100,7 +100,7 @@ trait CommonCorrelate {
|}
|""".stripMargin
} else if (joinType != JoinRelType.INNER) {
- throw new TableException(s"Unsupported SemiJoinType: $joinType for
correlate join.")
+ throw new TableException(s"Unsupported JoinRelType: $joinType for
correlate join.")
}
functionGenerator.generateFunction(
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonJoin.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonJoin.scala
index 3d98a4d..753ec40 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonJoin.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonJoin.scala
@@ -45,6 +45,8 @@ trait CommonJoin {
case JoinRelType.LEFT=> "LeftOuterJoin"
case JoinRelType.RIGHT => "RightOuterJoin"
case JoinRelType.FULL => "FullOuterJoin"
+ case JoinRelType.SEMI => "SemiJoin"
+ case JoinRelType.ANTI => "AntiJoin"
}
}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
index 1df75e6..2e319f4 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
@@ -199,6 +199,7 @@ class DataSetJoin(
rightKeys.toArray,
returnType,
config)
+ case _ => throw new TableException(s"$joinType is not supported.")
}
}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoinToCoProcessTranslator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoinToCoProcessTranslator.scala
index 846e452..8c418a3 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoinToCoProcessTranslator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoinToCoProcessTranslator.scala
@@ -18,13 +18,11 @@
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
-import org.apache.calcite.rex.{RexBuilder, RexNode}
import org.apache.flink.api.common.functions.FlatJoinFunction
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator
import org.apache.flink.streaming.api.operators.co.LegacyKeyedCoProcessOperator
-import org.apache.flink.table.api.{StreamQueryConfig, TableConfig}
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
ValidationException}
import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.runtime.CRowKeySelector
@@ -32,6 +30,9 @@ import org.apache.flink.table.runtime.join._
import org.apache.flink.table.runtime.types.CRow
import org.apache.flink.types.Row
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+impo
