Merge branch 'master' into fix-#1334_orc_timestamp

DTStack · Nov 6, 2022 · 2327b6f · 2327b6f
2 parents 6da4488 + 9ece5fb
commit 2327b6f
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 1 deletion.
diff --git a/...tor-ftp/src/main/java/com/dtstack/chunjun/connector/ftp/table/FtpDynamicTableFactory.java b/...tor-ftp/src/main/java/com/dtstack/chunjun/connector/ftp/table/FtpDynamicTableFactory.java
@@ -18,6 +18,7 @@
 
 package com.dtstack.chunjun.connector.ftp.table;
 
+import com.dtstack.chunjun.conf.FieldConf;
 import com.dtstack.chunjun.connector.ftp.conf.FtpConfig;
 import com.dtstack.chunjun.connector.ftp.options.FtpOptions;
 import com.dtstack.chunjun.connector.ftp.sink.FtpDynamicTableSink;
@@ -38,9 +39,12 @@
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.utils.TableSchemaUtils;
 
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -94,7 +98,18 @@ public DynamicTableSource createDynamicTableSource(Context context) {
  helper.discoverDecodingFormat(
  DeserializationFormatFactory.class, FtpOptions.FORMAT);
 
+ RowType rowType = (RowType) physicalSchema.toRowDataType().getLogicalType();
  FtpConfig ftpConfig = getFtpConfByOptions(config);
+ String[] fieldNames = physicalSchema.getFieldNames();
+ List<FieldConf> columnList = new ArrayList<>(fieldNames.length);
+ for (int i = 0; i < fieldNames.length; i++) {
+ FieldConf field = new FieldConf();
+ field.setName(fieldNames[i]);
+ field.setType(rowType.getTypeAt(i).asSummaryString());
+ field.setIndex(i);
+ columnList.add(field);
+ }
+ ftpConfig.setColumn(columnList);
 
  return new FtpDynamicTableSource(physicalSchema, ftpConfig, decodingFormat);
  }

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/Main.java b/chunjun-core/src/main/java/com/dtstack/chunjun/Main.java
@@ -28,6 +28,7 @@
 import com.dtstack.chunjun.constants.ConstantValue;
 import com.dtstack.chunjun.dirty.DirtyConf;
 import com.dtstack.chunjun.dirty.utils.DirtyConfUtil;
+import com.dtstack.chunjun.enums.ClusterMode;
 import com.dtstack.chunjun.enums.EJobType;
 import com.dtstack.chunjun.environment.EnvFactory;
 import com.dtstack.chunjun.environment.MyLocalStreamEnvironment;
@@ -321,8 +322,13 @@ private static void configStreamExecutionEnvironment(
  factoryHelper.setPluginLoadMode(options.getPluginLoadMode());
  factoryHelper.setEnv(env);
  factoryHelper.setExecutionMode(options.getMode());
-
  DirtyConf dirtyConf = DirtyConfUtil.parse(options);
+ // 注册core包
+ if (ClusterMode.local.name().equalsIgnoreCase(options.getMode())) {
+ factoryHelper.registerCachedFile(
+ "", Thread.currentThread().getContextClassLoader(), "");
+ }
+
  factoryHelper.registerCachedFile(
  dirtyConf.getType(),
  Thread.currentThread().getContextClassLoader(),