[1/2] flink git commit: [hotfix] [rpc] Add RpcConnectionTest to validate that connection buildup fails fast when endpoint is unreachable.
Repository: flink Updated Branches: refs/heads/flip-6 fdeda082f -> 3cda59339 [hotfix] [rpc] Add RpcConnectionTest to validate that connection buildup fails fast when endpoint is unreachable. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/629078ee Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/629078ee Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/629078ee Branch: refs/heads/flip-6 Commit: 629078ee3f5fcecd5498a81abaf8c99f9e614b02 Parents: fdeda08 Author: Stephan Ewen Authored: Wed Sep 21 13:03:17 2016 +0200 Committer: Stephan Ewen Committed: Fri Sep 23 19:44:13 2016 +0200 -- .../flink/runtime/rpc/AsyncCallsTest.java | 4 +- .../flink/runtime/rpc/RpcCompletenessTest.java | 14 ++-- .../flink/runtime/rpc/RpcConnectionTest.java| 86 3 files changed, 96 insertions(+), 8 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/629078ee/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java -- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java index e8255d4..7affdb9 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java @@ -43,9 +43,9 @@ public class AsyncCallsTest extends TestLogger { // shared test members // - private static ActorSystem actorSystem = AkkaUtils.createDefaultActorSystem(); + private static final ActorSystem actorSystem = AkkaUtils.createDefaultActorSystem(); - private static AkkaRpcService akkaRpcService = + private static final AkkaRpcService akkaRpcService = new AkkaRpcService(actorSystem, Time.milliseconds(1L)); @AfterClass http://git-wip-us.apache.org/repos/asf/flink/blob/629078ee/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java -- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java index ee3f784..53355e8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java @@ -30,6 +30,7 @@ import org.reflections.Reflections; import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -69,7 +70,8 @@ public class RpcCompletenessTest extends TestLogger { @SuppressWarnings("rawtypes") private void checkCompleteness(Class rpcEndpoint, Class rpcGateway) { - Method[] gatewayMethods = getRpcMethodsFromGateway(rpcGateway).toArray(new Method[0]); + List rpcMethodsFromGateway = getRpcMethodsFromGateway(rpcGateway); + Method[] gatewayMethods = rpcMethodsFromGateway.toArray(new Method[rpcMethodsFromGateway.size()]); Method[] serverMethods = rpcEndpoint.getMethods(); Map> rpcMethods = new HashMap<>(); @@ -360,13 +362,13 @@ public class RpcCompletenessTest extends TestLogger { } // Get all methods declared in current interface - for(Method method : interfaceClass.getDeclaredMethods()) { - allMethods.add(method); - } + Collections.addAll(allMethods, interfaceClass.getDeclaredMethods()); // Get all method inherited from super interface - for(Class superClass : interfaceClass.getInterfaces()) { - allMethods.addAll(getRpcMethodsFromGateway(superClass)); + for (Class superClass : interfaceClass.getInterfaces()) { + @SuppressWarnings("unchecked") + Class gatewayClass = (Class) superClass; + allMethods.addAll(getRpcMethodsFromGateway(gatewayClass)); } return allMethods; } http://git-wip-us.apache.org/repos/asf/flink/blob/629078ee/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java -- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java b/flink-runtime/src/te
[2/2] flink git commit: [FLINK-4408] [JobManager] Introduce JobMasterRunner and implement job submission & setting up the ExecutionGraph
[FLINK-4408] [JobManager] Introduce JobMasterRunner and implement job submission & setting up the ExecutionGraph This closes #2480 Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3cda5933 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3cda5933 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3cda5933 Branch: refs/heads/flip-6 Commit: 3cda59339ed593b04be3a897bf04a01e3673db5b Parents: 629078e Author: Kurt Young Authored: Thu Sep 8 12:00:13 2016 +0800 Committer: Stephan Ewen Committed: Fri Sep 23 20:30:05 2016 +0200 -- .../flink/api/common/JobExecutionResult.java| 2 +- .../flink/api/common/JobSubmissionResult.java | 2 +- .../HighAvailabilityServices.java | 12 + .../runtime/highavailability/NonHaServices.java | 16 +- .../runtime/jobmanager/OnCompletionActions.java | 31 ++ .../runtime/jobmanager/scheduler/Scheduler.java | 9 + .../runtime/jobmaster/JobManagerRunner.java | 288 +++ .../runtime/jobmaster/JobManagerServices.java | 73 +++ .../flink/runtime/jobmaster/JobMaster.java | 485 ++- .../runtime/jobmaster/JobMasterGateway.java | 13 + .../jobmaster/MiniClusterJobDispatcher.java | 385 +++ .../flink/runtime/rpc/FatalErrorHandler.java| 24 + .../runtime/taskexecutor/TaskExecutor.java | 12 + .../TestingHighAvailabilityServices.java| 39 +- .../jobmaster/JobManagerRunnerMockTest.java | 254 ++ .../flink/runtime/rpc/RpcConnectionTest.java| 17 +- 16 files changed, 1533 insertions(+), 129 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/3cda5933/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java -- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java b/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java index cb4ecc5..7286cc5 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java @@ -32,7 +32,7 @@ import java.util.concurrent.TimeUnit; @Public public class JobExecutionResult extends JobSubmissionResult { - private long netRuntime; + private final long netRuntime; private final Map accumulatorResults; http://git-wip-us.apache.org/repos/asf/flink/blob/3cda5933/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java -- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java b/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java index c5dc869..b0e7e24 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java @@ -26,7 +26,7 @@ import org.apache.flink.annotation.Public; @Public public class JobSubmissionResult { - private JobID jobID; + private final JobID jobID; public JobSubmissionResult(JobID jobID) { this.jobID = jobID; http://git-wip-us.apache.org/repos/asf/flink/blob/3cda5933/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java index 7634176..d67e927 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java @@ -19,6 +19,8 @@ package org.apache.flink.runtime.highavailability; import org.apache.flink.api.common.JobID; +import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory; +import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore; import org.apache.flink.runtime.leaderelection.LeaderElectionService; import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService; @@ -61,4 +63,14 @@ public interface HighAvailabilityServices { * @param jobID The identifier of the job running the election. */ LeaderElectionService getJobMasterLeaderElectionService(JobID jobID) throws Exception; + + /** +* Gets the checkpoint recovery factory for the job manager +*/ + CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception; + + /** +* Gets
flink git commit: [FLINK-4672] [taskmanager] Do not decorate Actor Kill messages
Repository: flink Updated Branches: refs/heads/release-1.1 62c666f57 -> caa0fbb21 [FLINK-4672] [taskmanager] Do not decorate Actor Kill messages Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/caa0fbb2 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/caa0fbb2 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/caa0fbb2 Branch: refs/heads/release-1.1 Commit: caa0fbb2157de56c9bdc4bbf8aedb73df90edede Parents: 62c666f Author: Stephan Ewen Authored: Fri Sep 23 18:42:47 2016 +0200 Committer: Stephan Ewen Committed: Mon Sep 26 11:25:01 2016 +0200 -- .../flink/runtime/taskmanager/TaskManager.scala | 2 +- .../runtime/taskmanager/TaskManagerTest.java| 33 ++-- 2 files changed, 31 insertions(+), 4 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/caa0fbb2/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala -- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala index a7dd789..8e787bb 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala @@ -1333,7 +1333,7 @@ class TaskManager( "\n" + "A fatal error occurred, forcing the TaskManager to shut down: " + message, cause) -self ! decorateMessage(Kill) +self ! Kill } override def notifyLeaderAddress(leaderAddress: String, leaderSessionID: UUID): Unit = { http://git-wip-us.apache.org/repos/asf/flink/blob/caa0fbb2/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java -- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java index ce88c09..1c50265 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java @@ -20,6 +20,7 @@ package org.apache.flink.runtime.taskmanager; import akka.actor.ActorRef; import akka.actor.ActorSystem; +import akka.actor.Kill; import akka.actor.Props; import akka.japi.Creator; import akka.testkit.JavaTestKit; @@ -55,6 +56,7 @@ import org.apache.flink.runtime.messages.StackTraceSampleMessages.ResponseStackT import org.apache.flink.runtime.messages.StackTraceSampleMessages.ResponseStackTraceSampleSuccess; import org.apache.flink.runtime.messages.StackTraceSampleMessages.TriggerStackTraceSample; import org.apache.flink.runtime.messages.TaskManagerMessages; +import org.apache.flink.runtime.messages.TaskManagerMessages.FatalError; import org.apache.flink.runtime.messages.TaskMessages; import org.apache.flink.runtime.messages.TaskMessages.CancelTask; import org.apache.flink.runtime.messages.TaskMessages.PartitionState; @@ -1369,6 +1371,28 @@ public class TaskManagerTest extends TestLogger { }}; } + @Test + public void testTerminationOnFatalError() { + new JavaTestKit(system){{ + + final ActorGateway taskManager = TestingUtils.createTaskManager( + system, + system.deadLetters(), // no jobmanager + new Configuration(), + true, + false); + + try { + watch(taskManager.actor()); + taskManager.tell(new FatalError("test fatal error", new Exception("something super bad"))); + expectTerminated(d, taskManager.actor()); + } + finally { + taskManager.tell(Kill.getInstance()); + } + }}; + } + // public static class SimpleJobManager extends FlinkUntypedActor { @@ -1549,11 +1573,14 @@ public class TaskManagerTest extends TestLogger { @Override public void invoke() throws Exception { - Object o = new Object(); + final Object o = new Object(); + //noinspection SynchronizationOnLocalVariableOrMethodParameter synchroni
flink git commit: [hotfix] Fix restart strategy class loading by using not lower cased class name
Repository: flink Updated Branches: refs/heads/master 4a8e94403 -> 28ff5a3c9 [hotfix] Fix restart strategy class loading by using not lower cased class name Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/28ff5a3c Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/28ff5a3c Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/28ff5a3c Branch: refs/heads/master Commit: 28ff5a3ce02b84f4e4aa568347ba5e13c3d2 Parents: 4a8e944 Author: Till Rohrmann Authored: Mon Sep 26 12:00:19 2016 +0200 Committer: Till Rohrmann Committed: Mon Sep 26 12:00:19 2016 +0200 -- .../runtime/executiongraph/restart/RestartStrategyFactory.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/28ff5a3c/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/RestartStrategyFactory.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/RestartStrategyFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/RestartStrategyFactory.java index ae92b3a..870bf63 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/RestartStrategyFactory.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/RestartStrategyFactory.java @@ -79,9 +79,9 @@ public abstract class RestartStrategyFactory implements Serializable { * @throws Exception which indicates that the RestartStrategy could not be instantiated. */ public static RestartStrategyFactory createRestartStrategyFactory(Configuration configuration) throws Exception { - String restartStrategyName = configuration.getString(ConfigConstants.RESTART_STRATEGY, "none").toLowerCase(); + String restartStrategyName = configuration.getString(ConfigConstants.RESTART_STRATEGY, "none"); - switch (restartStrategyName) { + switch (restartStrategyName.toLowerCase()) { case "none": // support deprecated ConfigConstants values final int numberExecutionRetries = configuration.getInteger(ConfigConstants.EXECUTION_RETRIES_KEY,
[2/2] flink git commit: [FLINK-4672] [taskmanager] Do not decorate Actor Kill messages
[FLINK-4672] [taskmanager] Do not decorate Actor Kill messages Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6f237cfe Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6f237cfe Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6f237cfe Branch: refs/heads/master Commit: 6f237cfe6f70b5b72fedd3dea6fbeb6c929631e8 Parents: 28ff5a3 Author: Stephan Ewen Authored: Fri Sep 23 18:42:47 2016 +0200 Committer: Stephan Ewen Committed: Mon Sep 26 14:11:05 2016 +0200 -- .../flink/runtime/taskmanager/TaskManager.scala | 2 +- .../runtime/taskmanager/TaskManagerTest.java| 33 ++-- 2 files changed, 31 insertions(+), 4 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/6f237cfe/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala -- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala index 9e2feb5..04f3168 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala @@ -1379,7 +1379,7 @@ class TaskManager( "\n" + "A fatal error occurred, forcing the TaskManager to shut down: " + message, cause) -self ! decorateMessage(Kill) +self ! Kill } override def notifyLeaderAddress(leaderAddress: String, leaderSessionID: UUID): Unit = { http://git-wip-us.apache.org/repos/asf/flink/blob/6f237cfe/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java -- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java index 0e53673..0774fd5 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java @@ -20,6 +20,7 @@ package org.apache.flink.runtime.taskmanager; import akka.actor.ActorRef; import akka.actor.ActorSystem; +import akka.actor.Kill; import akka.actor.Props; import akka.japi.Creator; import akka.testkit.JavaTestKit; @@ -55,6 +56,7 @@ import org.apache.flink.runtime.messages.StackTraceSampleMessages.ResponseStackT import org.apache.flink.runtime.messages.StackTraceSampleMessages.ResponseStackTraceSampleSuccess; import org.apache.flink.runtime.messages.StackTraceSampleMessages.TriggerStackTraceSample; import org.apache.flink.runtime.messages.TaskManagerMessages; +import org.apache.flink.runtime.messages.TaskManagerMessages.FatalError; import org.apache.flink.runtime.messages.TaskMessages; import org.apache.flink.runtime.messages.TaskMessages.CancelTask; import org.apache.flink.runtime.messages.TaskMessages.PartitionState; @@ -1367,6 +1369,28 @@ public class TaskManagerTest extends TestLogger { }}; } + @Test + public void testTerminationOnFatalError() { + new JavaTestKit(system){{ + + final ActorGateway taskManager = TestingUtils.createTaskManager( + system, + system.deadLetters(), // no jobmanager + new Configuration(), + true, + false); + + try { + watch(taskManager.actor()); + taskManager.tell(new FatalError("test fatal error", new Exception("something super bad"))); + expectTerminated(d, taskManager.actor()); + } + finally { + taskManager.tell(Kill.getInstance()); + } + }}; + } + // public static class SimpleJobManager extends FlinkUntypedActor { @@ -1547,11 +1571,14 @@ public class TaskManagerTest extends TestLogger { @Override public void invoke() throws Exception { - Object o = new Object(); + final Object o = new Object(); + //noinspection SynchronizationOnLocalVariableOrMethodParameter synchronized (o) { - o.wait(); + //noins
[1/2] flink git commit: [FLINK-4218] [checkpoints] Do not rely on FileSystem to determing state sizes
Repository: flink Updated Branches: refs/heads/master 28ff5a3c9 -> 95e9004e3 [FLINK-4218] [checkpoints] Do not rely on FileSystem to determing state sizes This prevents failures on eventually consistent S3, where the operations for keys (=entries in the parent directory/bucket) are not guaranteed to be immediately consistent (visible) after a blob was written. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/95e9004e Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/95e9004e Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/95e9004e Branch: refs/heads/master Commit: 95e9004e36fffae755eab7aa3d5d0d5e8bfb7113 Parents: 6f237cf Author: Stephan Ewen Authored: Fri Sep 23 15:16:27 2016 +0200 Committer: Stephan Ewen Committed: Mon Sep 26 14:11:05 2016 +0200 -- .../runtime/checkpoint/CompletedCheckpoint.java | 2 +- .../flink/runtime/checkpoint/TaskState.java | 2 +- .../savepoint/SavepointV1Serializer.java| 6 ++-- .../flink/runtime/state/ChainedStateHandle.java | 2 +- .../runtime/state/KeyGroupsStateHandle.java | 2 +- .../state/RetrievableStreamStateHandle.java | 9 +++--- .../apache/flink/runtime/state/StateObject.java | 6 ++-- .../state/filesystem/FileStateHandle.java | 32 .../filesystem/FsCheckpointStreamFactory.java | 9 +- .../FileSystemStateStorageHelper.java | 15 +++-- ...ZooKeeperCompletedCheckpointStoreITCase.java | 2 +- .../stats/SimpleCheckpointStatsTrackerTest.java | 18 +++ .../state/AbstractCloseableHandleTest.java | 6 ++-- .../tasks/InterruptSensitiveRestoreTest.java| 3 +- 14 files changed, 58 insertions(+), 56 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/95e9004e/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java index e412006..7cb3916 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java @@ -108,7 +108,7 @@ public class CompletedCheckpoint implements StateObject { } @Override - public long getStateSize() throws Exception { + public long getStateSize() throws IOException { long result = 0L; for (TaskState taskState : taskStates.values()) { http://git-wip-us.apache.org/repos/asf/flink/blob/95e9004e/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java index 9025090..657dd60 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java @@ -152,7 +152,7 @@ public class TaskState implements StateObject { @Override - public long getStateSize() throws Exception { + public long getStateSize() throws IOException { long result = 0L; for (int i = 0; i < parallelism; i++) { http://git-wip-us.apache.org/repos/asf/flink/blob/95e9004e/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java index 8e05b81..f07f44f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java @@ -197,6 +197,7 @@ class SavepointV1Serializer implements SavepointSerializer { } else if (stateHandle instanceof FileStateHandle) { dos.writeByte(FILE_STREAM_STATE_HANDLE); FileStateHandle fileStateHandle = (FileStateHandle) stateHandle; + dos.writeLong(stateHandle.getStateSize()); dos.writeUTF(fileStateHandle.getFilePath().toString()); } else if (stateHandle instanceof ByteStreamStateHandle) { @@ -218,12 +219,13 @@ class Savepoint
flink git commit: [FLINK-4580] [rpc] Report rpc invocation exceptions to the caller
Repository: flink Updated Branches: refs/heads/flip-6 3cda59339 -> 2a61e74b9 [FLINK-4580] [rpc] Report rpc invocation exceptions to the caller This closes #2526. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2a61e74b Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2a61e74b Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2a61e74b Branch: refs/heads/flip-6 Commit: 2a61e74b9108835ffb5f8ab89d67cb7105801594 Parents: 3cda593 Author: Till Rohrmann Authored: Wed Sep 21 15:18:27 2016 +0200 Committer: Till Rohrmann Committed: Mon Sep 26 14:26:54 2016 +0200 -- .../flink/runtime/rpc/akka/AkkaRpcActor.java| 53 .../runtime/rpc/akka/AkkaRpcActorTest.java | 34 + 2 files changed, 66 insertions(+), 21 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/2a61e74b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java index 59daa46..1b456a7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java @@ -35,6 +35,7 @@ import org.apache.flink.runtime.rpc.akka.messages.Processing; import org.apache.flink.runtime.rpc.akka.messages.RpcInvocation; import org.apache.flink.runtime.rpc.akka.messages.RunAsync; +import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -86,11 +87,11 @@ class AkkaRpcActor> extends Untyp unstashAll(); getContext().become(new Procedure() { @Override - public void apply(Object message) throws Exception { - if (message.equals(Processing.STOP)) { + public void apply(Object msg) throws Exception { + if (msg.equals(Processing.STOP)) { getContext().unbecome(); } else { - handleMessage(message); + handleMessage(msg); } } }); @@ -130,21 +131,36 @@ class AkkaRpcActor> extends Untyp * @param rpcInvocation Rpc invocation message */ private void handleRpcInvocation(RpcInvocation rpcInvocation) { + Method rpcMethod = null; + try { String methodName = rpcInvocation.getMethodName(); Class[] parameterTypes = rpcInvocation.getParameterTypes(); - Method rpcMethod = lookupRpcMethod(methodName, parameterTypes); + rpcMethod = lookupRpcMethod(methodName, parameterTypes); + } catch(ClassNotFoundException e) { + LOG.error("Could not load method arguments.", e); + + RpcConnectionException rpcException = new RpcConnectionException("Could not load method arguments.", e); + getSender().tell(new Status.Failure(rpcException), getSelf()); + } catch (IOException e) { + LOG.error("Could not deserialize rpc invocation message.", e); + + RpcConnectionException rpcException = new RpcConnectionException("Could not deserialize rpc invocation message.", e); + getSender().tell(new Status.Failure(rpcException), getSelf()); + } catch (final NoSuchMethodException e) { + LOG.error("Could not find rpc method for rpc invocation.", e); + + RpcConnectionException rpcException = new RpcConnectionException("Could not find rpc method for rpc invocation.", e); + getSender().tell(new Status.Failure(rpcException), getSelf()); + } - if (rpcMethod.getReturnType().equals(Void.TYPE)) { - // No return value to send back - try { + if (rpcMethod != null) { + try { + if (rpcMethod.getReturnType().equals(Void.TYPE)) { + // No return value to send back rpcMethod.invoke(rpcEndpoint, rpcInvocation.g
flink git commit: [FLINK-4662] Bump Calcite version up to 1.9
Repository: flink Updated Branches: refs/heads/master 95e9004e3 -> 8fa313c39 [FLINK-4662] Bump Calcite version up to 1.9 This closes #2535. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8fa313c3 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8fa313c3 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8fa313c3 Branch: refs/heads/master Commit: 8fa313c39fbad7bb96327477544d1ec15e8dc0f6 Parents: 95e9004 Author: Jark Wu Authored: Thu Sep 22 21:47:37 2016 +0800 Committer: twalthr Committed: Mon Sep 26 15:58:39 2016 +0200 -- flink-libraries/flink-table/pom.xml | 2 +- .../apache/flink/api/table/FlinkPlannerImpl.scala| 15 --- .../apache/flink/api/table/FlinkTypeFactory.scala| 8 .../flink/api/table/codegen/CodeGenerator.scala | 4 ++-- .../flink/api/table/expressions/arithmetic.scala | 5 + .../table/plan/nodes/dataset/DataSetConvention.scala | 7 ++- .../api/table/plan/nodes/dataset/DataSetRel.scala| 4 ++-- .../plan/nodes/datastream/DataStreamConvention.scala | 7 ++- 8 files changed, 34 insertions(+), 18 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/8fa313c3/flink-libraries/flink-table/pom.xml -- diff --git a/flink-libraries/flink-table/pom.xml b/flink-libraries/flink-table/pom.xml index e83a778..4c91f1c 100644 --- a/flink-libraries/flink-table/pom.xml +++ b/flink-libraries/flink-table/pom.xml @@ -51,7 +51,7 @@ under the License. org.apache.calcite calcite-core - 1.7.0 + 1.9.0 org.apache.calcite.avatica http://git-wip-us.apache.org/repos/asf/flink/blob/8fa313c3/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala index bb08654..97e5cf2 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala @@ -98,10 +98,10 @@ class FlinkPlannerImpl( assert(validatedSqlNode != null) val rexBuilder: RexBuilder = createRexBuilder val cluster: RelOptCluster = RelOptCluster.create(planner, rexBuilder) + val config = SqlToRelConverter.configBuilder() +.withTrimUnusedFields(false).withConvertTableAccess(false).build() val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter( -new ViewExpanderImpl, validator, createCatalogReader, cluster, convertletTable) - sqlToRelConverter.setTrimUnusedFields(false) - sqlToRelConverter.enableTableAccessConversion(false) +new ViewExpanderImpl, validator, createCatalogReader, cluster, convertletTable, config) root = sqlToRelConverter.convertQuery(validatedSqlNode, false, true) root = root.withRel(sqlToRelConverter.flattenTypes(root.rel, true)) root = root.withRel(RelDecorrelator.decorrelateQuery(root.rel)) @@ -118,7 +118,8 @@ class FlinkPlannerImpl( override def expandView( rowType: RelDataType, queryString: String, -schemaPath: util.List[String]): RelRoot = { +schemaPath: util.List[String], +viewPath: util.List[String]): RelRoot = { val parser: SqlParser = SqlParser.create(queryString, parserConfig) var sqlNode: SqlNode = null @@ -136,10 +137,10 @@ class FlinkPlannerImpl( val validatedSqlNode: SqlNode = validator.validate(sqlNode) val rexBuilder: RexBuilder = createRexBuilder val cluster: RelOptCluster = RelOptCluster.create(planner, rexBuilder) + val config: SqlToRelConverter.Config = SqlToRelConverter.configBuilder +.withTrimUnusedFields(false).withConvertTableAccess(false).build val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter( -new ViewExpanderImpl, validator, catalogReader, cluster, convertletTable) - sqlToRelConverter.setTrimUnusedFields(false) - sqlToRelConverter.enableTableAccessConversion(false) +new ViewExpanderImpl, validator, catalogReader, cluster, convertletTable, config) root = sqlToRelConverter.convertQuery(validatedSqlNode, true, false) root = root.withRel(sqlToRelConverter.flattenTypes(root.rel, true)) root = root.withRel(RelDecorrelator.decorrelateQuery(root.re
flink git commit: [FLINK-4684] [checkpoints] Remove redundant class loader from CheckpointCoordinator
Repository: flink Updated Branches: refs/heads/master 8fa313c39 -> 70e71c161 [FLINK-4684] [checkpoints] Remove redundant class loader from CheckpointCoordinator Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/70e71c16 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/70e71c16 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/70e71c16 Branch: refs/heads/master Commit: 70e71c16177b40c2418e6a8ca0838bf117f6a926 Parents: 8fa313c Author: Stephan Ewen Authored: Mon Sep 26 12:32:10 2016 +0200 Committer: Stephan Ewen Committed: Mon Sep 26 18:05:01 2016 +0200 -- .../checkpoint/CheckpointCoordinator.java | 5 - .../flink/runtime/checkpoint/TaskState.java | 2 +- .../runtime/executiongraph/ExecutionGraph.java | 1 - .../apache/flink/runtime/state/StateUtil.java | 2 +- .../checkpoint/CheckpointCoordinatorTest.java | 101 +++ .../checkpoint/CheckpointStateRestoreTest.java | 7 +- 6 files changed, 38 insertions(+), 80 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/70e71c16/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java index fc40911..6a43ddf 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java @@ -112,9 +112,6 @@ public class CheckpointCoordinator { * need to be ascending across job managers. */ private final CheckpointIDCounter checkpointIdCounter; - /** Class loader used to deserialize the state handles (as they may be user-defined) */ - private final ClassLoader userClassLoader; - /** The base checkpoint interval. Actual trigger time may be affected by the * max concurrent checkpoints and minimum-pause values */ private final long baseInterval; @@ -167,7 +164,6 @@ public class CheckpointCoordinator { ExecutionVertex[] tasksToTrigger, ExecutionVertex[] tasksToWaitFor, ExecutionVertex[] tasksToCommitTo, - ClassLoader userClassLoader, CheckpointIDCounter checkpointIDCounter, CompletedCheckpointStore completedCheckpointStore, SavepointStore savepointStore, @@ -198,7 +194,6 @@ public class CheckpointCoordinator { this.completedCheckpointStore = checkNotNull(completedCheckpointStore); this.savepointStore = checkNotNull(savepointStore); this.recentPendingCheckpoints = new ArrayDeque<>(NUM_GHOST_CHECKPOINT_IDS); - this.userClassLoader = checkNotNull(userClassLoader); this.statsTracker = checkNotNull(statsTracker); this.timer = new Timer("Checkpoint Timer", true); http://git-wip-us.apache.org/repos/asf/flink/blob/70e71c16/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java index 657dd60..f5e3618 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java @@ -62,7 +62,7 @@ public class TaskState implements StateObject { "Parallelism " + parallelism + " is not smaller or equal to max parallelism " + maxParallelism + "."); this.jobVertexID = jobVertexID; - //preallocate lists of the required size, so that we can randomly set values to indexes + this.subtaskStates = new HashMap<>(parallelism); this.keyGroupsStateHandles = new HashMap<>(parallelism); http://git-wip-us.apache.org/repos/asf/flink/blob/70e71c16/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java -- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java index c3cf297..7c3fa0b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java +++ b/f
flink git commit: [FLINK-4590] [table] Some Table API tests are failing when debug lvl is set to DEBUG
Repository: flink Updated Branches: refs/heads/master 70e71c161 -> 7eb45c133 [FLINK-4590] [table] Some Table API tests are failing when debug lvl is set to DEBUG This closes #2504. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7eb45c13 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7eb45c13 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7eb45c13 Branch: refs/heads/master Commit: 7eb45c133c49933b14719f06bf68ccf162a3e0b2 Parents: 70e71c1 Author: twalthr Authored: Fri Sep 16 10:52:28 2016 +0200 Committer: twalthr Committed: Mon Sep 26 18:48:47 2016 +0200 -- .../table/plan/nodes/dataset/BatchScan.scala| 10 ++--- .../nodes/dataset/BatchTableSourceScan.scala| 6 +-- .../plan/nodes/dataset/DataSetAggregate.scala | 28 +++--- .../table/plan/nodes/dataset/DataSetCalc.scala | 16 .../plan/nodes/dataset/DataSetIntersect.scala | 16 .../table/plan/nodes/dataset/DataSetJoin.scala | 20 +- .../table/plan/nodes/dataset/DataSetMinus.scala | 39 +--- .../table/plan/nodes/dataset/DataSetScan.scala | 6 +-- .../table/plan/nodes/dataset/DataSetSort.scala | 22 ++- .../table/plan/nodes/dataset/DataSetUnion.scala | 19 ++ .../plan/nodes/dataset/DataSetValues.scala | 16 .../plan/nodes/datastream/DataStreamCalc.scala | 16 .../plan/nodes/datastream/DataStreamScan.scala | 6 +-- .../plan/nodes/datastream/DataStreamUnion.scala | 16 .../nodes/datastream/DataStreamValues.scala | 12 +++--- .../plan/nodes/datastream/StreamScan.scala | 8 ++-- .../src/test/resources/log4j-test.properties| 27 ++ .../src/test/resources/logback-test.xml | 29 +++ 18 files changed, 193 insertions(+), 119 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/7eb45c13/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchScan.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchScan.scala index 85ed6ef..15b2081 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchScan.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchScan.scala @@ -36,14 +36,14 @@ abstract class BatchScan( cluster: RelOptCluster, traitSet: RelTraitSet, table: RelOptTable, -rowType: RelDataType) +rowRelDataType: RelDataType) extends TableScan(cluster, traitSet, table) with DataSetRel { - override def deriveRowType() = rowType + override def deriveRowType() = rowRelDataType override def toString: String = { -s"Source(from: (${rowType.getFieldNames.asScala.toList.mkString(", ")}))" +s"Source(from: (${getRowType.getFieldNames.asScala.toList.mkString(", ")}))" } override def computeSelfCost (planner: RelOptPlanner, metadata: RelMetadataQuery): RelOptCost = { @@ -81,14 +81,14 @@ abstract class BatchScan( val mapFunc = getConversionMapper( config, -false, +nullableInput = false, inputType, determinedType, "DataSetSourceConversion", getRowType.getFieldNames, Some(flinkTable.fieldIndexes)) - val opName = s"from: (${rowType.getFieldNames.asScala.toList.mkString(", ")})" + val opName = s"from: (${getRowType.getFieldNames.asScala.toList.mkString(", ")})" input.map(mapFunc).name(opName) } http://git-wip-us.apache.org/repos/asf/flink/blob/7eb45c13/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchTableSourceScan.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchTableSourceScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchTableSourceScan.scala index 027a5be..10d9534 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchTableSourceScan.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/BatchTableSourceScan.scala @@ -35,15 +35,15 @@ class BatchTableSourceScan( rowType: RelDataType) extends BatchScan(cluster, traitSet, table, rowType) { - val tableSourceTable = table.unwrap(classOf[TableSourceTable]) + val tableSourceTable = getTable.unwrap(classOf[
flink git commit: [FLINK-4252] [table] Validate input and output classes of Table API
Repository: flink Updated Branches: refs/heads/master 7eb45c133 -> f150f9877 [FLINK-4252] [table] Validate input and output classes of Table API This closes #2507. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f150f987 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f150f987 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f150f987 Branch: refs/heads/master Commit: f150f987772c8d96f41a5acd1d20cba6622cb5c9 Parents: 7eb45c1 Author: twalthr Authored: Fri Sep 16 14:27:47 2016 +0200 Committer: twalthr Committed: Mon Sep 26 18:58:44 2016 +0200 -- .../api/java/table/BatchTableEnvironment.scala | 5 ++--- .../flink/api/table/BatchTableEnvironment.scala | 4 +++- .../flink/api/table/FlinkRelBuilder.scala | 4 ++-- .../api/table/StreamTableEnvironment.scala | 2 ++ .../flink/api/table/TableEnvironment.scala | 15 + .../api/java/batch/table/FromDataSetITCase.java | 23 6 files changed, 47 insertions(+), 6 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/f150f987/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/BatchTableEnvironment.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/BatchTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/BatchTableEnvironment.scala index 9ba5b20..a4f40d5 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/BatchTableEnvironment.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/BatchTableEnvironment.scala @@ -17,12 +17,11 @@ */ package org.apache.flink.api.java.table -import org.apache.flink.api.common.io.InputFormat import org.apache.flink.api.common.typeinfo.TypeInformation -import org.apache.flink.api.java.{ExecutionEnvironment, DataSet} import org.apache.flink.api.java.typeutils.TypeExtractor +import org.apache.flink.api.java.{DataSet, ExecutionEnvironment} import org.apache.flink.api.table.expressions.ExpressionParser -import org.apache.flink.api.table.{Row, TableConfig, Table} +import org.apache.flink.api.table.{Table, TableConfig} /** * The [[org.apache.flink.api.table.TableEnvironment]] for a Java batch [[DataSet]] http://git-wip-us.apache.org/repos/asf/flink/blob/f150f987/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/BatchTableEnvironment.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/BatchTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/BatchTableEnvironment.scala index eb4c819..ad3ff7a 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/BatchTableEnvironment.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/BatchTableEnvironment.scala @@ -190,7 +190,7 @@ abstract class BatchTableEnvironment( * * @param table The table for which the AST and execution plan will be returned. */ - def explain(table: Table): String = explain(table: Table, false) + def explain(table: Table): String = explain(table: Table, extended = false) /** * Registers a [[DataSet]] as a table under a given name in the [[TableEnvironment]]'s catalog. @@ -240,6 +240,8 @@ abstract class BatchTableEnvironment( */ protected def translate[A](table: Table)(implicit tpe: TypeInformation[A]): DataSet[A] = { +validateType(tpe) + val relNode = table.getRelNode // decorrelate http://git-wip-us.apache.org/repos/asf/flink/blob/f150f987/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala index e3bb97e..3827f05 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala @@ -19,7 +19,7 @@ package org.apache.flink.api.table import org.apache.calcite.jdbc.CalciteSchema -import org.apache.calcite.plan.{Context, RelOptCluster, RelOptSchema} +import org.apache.calcite.plan.{Context, RelOptCluster, RelOptPlanner, RelOptSchema} import org.apache.calcite.prepare.CalciteCatalogReader import org.apache.calcite.rex.RexBuilder import org.apache.calcite.schema.SchemaPlus @@ -38,7 +38,7 @@ class FlinkR
flink git commit: [FLINK-4241] [table] Cryptic expression parser exceptions
Repository: flink Updated Branches: refs/heads/master f150f9877 -> ef1598498 [FLINK-4241] [table] Cryptic expression parser exceptions This closes #2529. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ef159849 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ef159849 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ef159849 Branch: refs/heads/master Commit: ef15984988e883ced5311b332e5e5d8521c9573f Parents: f150f98 Author: twalthr Authored: Wed Sep 21 17:12:31 2016 +0200 Committer: twalthr Committed: Mon Sep 26 19:06:43 2016 +0200 -- docs/dev/table_api.md | 2 +- .../table/expressions/ExpressionParser.scala| 56 2 files changed, 34 insertions(+), 24 deletions(-) -- http://git-wip-us.apache.org/repos/asf/flink/blob/ef159849/docs/dev/table_api.md -- diff --git a/docs/dev/table_api.md b/docs/dev/table_api.md index bb083ff..1d03b38 100644 --- a/docs/dev/table_api.md +++ b/docs/dev/table_api.md @@ -953,7 +953,7 @@ alias = logic | ( logic , "AS" , fieldReference ) ; logic = comparison , [ ( "&&" | "||" ) , comparison ] ; -comparison = term , [ ( "=" | "===" | "!=" | "!==" | ">" | ">=" | "<" | "<=" ) , term ] ; +comparison = term , [ ( "=" | "==" | "===" | "!=" | "!==" | ">" | ">=" | "<" | "<=" ) , term ] ; term = product , [ ( "+" | "-" ) , product ] ; http://git-wip-us.apache.org/repos/asf/flink/blob/ef159849/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala -- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala index ae027e9..4707adf 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala @@ -154,9 +154,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers { } lazy val literalExpr: PackratParser[Expression] = -numberLiteral | - stringLiteralFlink | singleQuoteStringLiteral | - boolLiteral | nullLiteral +numberLiteral | stringLiteralFlink | singleQuoteStringLiteral | boolLiteral | nullLiteral lazy val fieldReference: PackratParser[NamedExpression] = (STAR | ident) ^^ { sym => UnresolvedFieldReference(sym) @@ -334,7 +332,8 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers { // suffix/prefix composite - lazy val composite: PackratParser[Expression] = suffixed | prefixed | atom + lazy val composite: PackratParser[Expression] = suffixed | prefixed | atom | +failure("Composite expression expected.") // unary ops @@ -342,22 +341,25 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers { lazy val unaryMinus: PackratParser[Expression] = "-" ~> composite ^^ { e => UnaryMinus(e) } - lazy val unary = composite | unaryNot | unaryMinus + lazy val unary = composite | unaryNot | unaryMinus | +failure("Unary expression expected.") // arithmetic lazy val product = unary * ( "*" ^^^ { (a:Expression, b:Expression) => Mul(a,b) } | - "/" ^^^ { (a:Expression, b:Expression) => Div(a,b) } | - "%" ^^^ { (a:Expression, b:Expression) => Mod(a,b) } ) +"/" ^^^ { (a:Expression, b:Expression) => Div(a,b) } | +"%" ^^^ { (a:Expression, b:Expression) => Mod(a,b) } ) | +failure("Product expected.") lazy val term = product * ( "+" ^^^ { (a:Expression, b:Expression) => Plus(a,b) } | - "-" ^^^ { (a:Expression, b:Expression) => Minus(a,b) } ) +"-" ^^^ { (a:Expression, b:Expression) => Minus(a,b) } ) | +failure("Term expected.") // Comparison - lazy val equalTo: PackratParser[Expression] = term ~ ("===" | "=") ~ term ^^ { + lazy val equalTo: PackratParser[Expression] = term ~ ("===" | "==" | "=") ~ term ^^ { case l ~ _ ~ r => EqualTo(l, r) } @@ -382,23 +384,26 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers { } lazy val comparison: PackratParser[Expression] = - equalTo | notEqualTo | - greaterThan | greaterThanOrEqual | - lessThan | lessThanOrEqual | term +equalTo | notEqualTo | +greaterThan | greaterThanOrEqual | +lessThan | lessThanOrEqual | term | +failure("Comparison expected.") // logic lazy val logic = comparison * ( "&&" ^^^ { (a:Expression, b:Expression) => And(a,b) } | - "||" ^^^ { (a:Expression, b:Expression) => Or(a,b) }
buildbot failure in on flink-docs-release-0.10
The Buildbot has detected a new failure on builder flink-docs-release-0.10 while building . Full details are available at: https://ci.apache.org/builders/flink-docs-release-0.10/builds/343 Buildbot URL: https://ci.apache.org/ Buildslave for this Build: bb_slave2_ubuntu Build Reason: The Nightly scheduler named 'flink-nightly-docs-release-0.10' triggered this build Build Source Stamp: [branch release-0.10] HEAD Blamelist: BUILD FAILED: failed compile Sincerely, -The Buildbot