You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2023/09/08 20:29:03 UTC
Change in asterixdb[master]: [WIP] Prepare for embedding values in parquet
From Wail Alkowaileet <wa...@gmail.com>:
Wail Alkowaileet has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17767 )
Change subject: [WIP] Prepare for embedding values in parquet
......................................................................
[WIP] Prepare for embedding values in parquet
Change-Id: Idde191ab6d99c48022bb65bec648e83e249e4b75
---
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/DCPConverterFactory.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSourceFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MapredParquetInputFormat.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/DelimitedDataParser.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IDataParser.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/IRecordConverterFactory.java
M asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/test/RecordWithMetaTest.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataRuntimeContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalStreamRuntimeDataContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVToRecordWithMetadataAndPKConverter.java
M asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/http/HttpServerRecordReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/TweetParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReaderFactory.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/DefaultExternalRuntimeDataContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/RecordWithMetadataParser.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RecordWithMetadataParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalReaderRuntimeDataContext.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetRecordReaderWrapper.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParser.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/abstracts/AbstractExternalInputStreamFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVWithRecordConverterFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureDataLakeParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/DelimitedDataParserFactory.java
M asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/factory/TestRecordWithPKParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RSSParserFactory.java
M asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/AbstractJsonDataParser.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/ADMDataParserFactory.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/NoOpExternalRuntimeDataContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/NoOpDataParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/JSONDataParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/AsterixInputStream.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IStreamDataParserFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/JSONDataParser.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/gcs/parquet/GCSParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/evaluators/StringJsonParseEval.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
53 files changed, 707 insertions(+), 391 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/67/17767/1
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/AsterixInputStream.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/AsterixInputStream.java
index 8eb4475..2ab6cd8 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/AsterixInputStream.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/AsterixInputStream.java
@@ -50,10 +50,6 @@
this.notificationHandler = notificationHandler;
}
- public void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
- this.valueEmbedder = valueEmbedder;
- }
-
public String getStreamName() {
return "";
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IDataParser.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IDataParser.java
index c1d6797..5dbc383 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IDataParser.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IDataParser.java
@@ -24,7 +24,6 @@
import org.apache.asterix.builders.OrderedListBuilder;
import org.apache.asterix.builders.RecordBuilder;
import org.apache.asterix.builders.UnorderedListBuilder;
-import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
import org.apache.asterix.om.base.AMutableOrderedList;
import org.apache.asterix.om.base.AMutableRecord;
@@ -39,9 +38,6 @@
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
public interface IDataParser {
- default void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
- // NoOp
- }
/*
* The following two static methods are expensive. right now, they are used by RSSFeeds and
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataRuntimeContext.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataRuntimeContext.java
new file mode 100644
index 0000000..00a9919
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataRuntimeContext.java
@@ -0,0 +1,41 @@
+/*
+ * 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.asterix.external.api;
+
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+
+/**
+ * A runtime context, where different {@link IRecordReader} and {@link IDataParser} can have different implementations
+ * of this interface. This context can be created by {@link IExternalDataSourceFactory}.
+ */
+public interface IExternalDataRuntimeContext {
+ IHyracksTaskContext getTaskContext();
+
+ int getPartition();
+
+ IExternalFilterValueEmbedder getValueEmbedder();
+
+ Supplier<String> getDatasourceNameSupplier();
+
+ LongSupplier getLineNumberSupplier();
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
index 058fe89..2cab786 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
@@ -28,11 +28,11 @@
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
-import org.apache.asterix.external.input.filter.NoOpFilterValueEmbedder;
-import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.asterix.external.provider.context.DefaultExternalRuntimeDataContext;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.IWarningCollector;
@@ -71,8 +71,9 @@
void configure(IServiceContext ctx, Map<String, String> configuration, IWarningCollector warningCollector,
IExternalFilterEvaluatorFactory filterEvaluatorFactory) throws AlgebricksException, HyracksDataException;
- default IExternalFilterValueEmbedder createFilterValueEmbedder(IWarningCollector warningCollector) {
- return NoOpFilterValueEmbedder.INSTANCE;
+ default IExternalDataRuntimeContext createExternalDataRuntimeContext(IHyracksTaskContext context, int partition)
+ throws HyracksDataException {
+ return new DefaultExternalRuntimeDataContext(context, partition);
}
/**
@@ -94,7 +95,8 @@
throw new AlgebricksException("Configurations don't have any stores");
}
int i = 0;
- outer: while (i < count) {
+ outer:
+ while (i < count) {
Iterator<String> storeIt = stores.iterator();
while (storeIt.hasNext()) {
String node = storeIt.next();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
index f544ca0..e4ab8bd 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
@@ -19,18 +19,8 @@
package org.apache.asterix.external.api;
import java.io.IOException;
-import java.util.function.LongSupplier;
@FunctionalInterface
public interface IRecordConverter<I, O> {
-
- public O convert(IRawRecord<? extends I> input) throws IOException;
-
- /**
- * Configures the converter with information suppliers from the {@link IRecordReader} data source.
- *
- * @param lineNumber line number supplier
- */
- default void configure(LongSupplier lineNumber) {
- }
+ O convert(IRawRecord<? extends I> input) throws IOException;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParser.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParser.java
index 31e1764..2cb4c4b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParser.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParser.java
@@ -19,8 +19,6 @@
package org.apache.asterix.external.api;
import java.io.DataOutput;
-import java.util.function.LongSupplier;
-import java.util.function.Supplier;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -37,14 +35,5 @@
* @return true if the record was parsed successfully and written to out. False, otherwise.
* @throws HyracksDataException HyracksDataException
*/
- public boolean parse(IRawRecord<? extends T> record, DataOutput out) throws HyracksDataException;
-
- /**
- * Configures the parser with information suppliers from the {@link IRecordReader} data source.
- *
- * @param dataSourceName data source name supplier
- * @param lineNumber line number supplier
- */
- default void configure(Supplier<String> dataSourceName, LongSupplier lineNumber) {
- }
+ boolean parse(IRawRecord<? extends T> record, DataOutput out) throws HyracksDataException;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParserFactory.java
index 2ddbbcd..b9032a0 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordDataParserFactory.java
@@ -19,16 +19,15 @@
package org.apache.asterix.external.api;
import org.apache.asterix.external.api.IExternalDataSourceFactory.DataSourceType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface IRecordDataParserFactory<T> extends IDataParserFactory {
- public IRecordDataParser<T> createRecordParser(IHyracksTaskContext ctx) throws HyracksDataException;
+ IRecordDataParser<T> createRecordParser(IExternalDataRuntimeContext context) throws HyracksDataException;
- public Class<?> getRecordClass();
+ Class<?> getRecordClass();
@Override
- public default DataSourceType getDataSourceType() {
+ default DataSourceType getDataSourceType() {
return DataSourceType.RECORDS;
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
index b7d4c85..b80e46c 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
@@ -25,7 +25,6 @@
import org.apache.asterix.active.message.ActiveManagerMessage;
import org.apache.asterix.external.dataflow.AbstractFeedDataFlowController;
-import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.IFeedLogManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -75,10 +74,6 @@
*/
public void setFeedLogManager(IFeedLogManager feedLogManager) throws HyracksDataException;
- default void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
- // NoOp
- }
-
/**
* gives the record reader a chance to recover from IO errors during feed intake
*/
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReaderFactory.java
index 8849508..64be02e 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReaderFactory.java
@@ -22,12 +22,11 @@
import java.util.Set;
import org.apache.asterix.external.util.ExternalDataConstants;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface IRecordReaderFactory<T> extends IExternalDataSourceFactory {
- IRecordReader<? extends T> createRecordReader(IHyracksTaskContext ctx, int partition) throws HyracksDataException;
+ IRecordReader<? extends T> createRecordReader(IExternalDataRuntimeContext context) throws HyracksDataException;
Class<?> getRecordClass();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IStreamDataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IStreamDataParserFactory.java
index ad9acc6..7b56319 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IStreamDataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IStreamDataParserFactory.java
@@ -18,11 +18,9 @@
*/
package org.apache.asterix.external.api;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface IStreamDataParserFactory extends IDataParserFactory {
- public IStreamDataParser createInputStreamParser(IHyracksTaskContext ctx, int partition)
- throws HyracksDataException;
+ IStreamDataParser createInputStreamParser(IExternalDataRuntimeContext context) throws HyracksDataException;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
index d7093b9..b010a4b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
@@ -18,9 +18,10 @@
*/
package org.apache.asterix.external.input;
-import static org.apache.asterix.external.util.ExternalDataConstants.CONTAINER_NAME_FIELD_NAME;
-import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_PARQUET;
-import static org.apache.hyracks.api.util.ExceptionUtils.getMessageOrToString;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
import java.io.IOException;
import java.util.Arrays;
@@ -33,21 +34,20 @@
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
import org.apache.asterix.external.api.AsterixInputStream;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.input.record.reader.hdfs.HDFSRecordReader;
import org.apache.asterix.external.input.record.reader.hdfs.parquet.ParquetFileRecordReader;
import org.apache.asterix.external.input.record.reader.stream.StreamRecordReader;
import org.apache.asterix.external.input.stream.HDFSInputStream;
import org.apache.asterix.external.provider.StreamRecordReaderProvider;
+import org.apache.asterix.external.provider.context.ExternalStreamRuntimeDataContext;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.external.util.HDFSUtils;
-import org.apache.hadoop.mapred.InputSplit;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.RecordReader;
-import org.apache.hadoop.mapred.Reporter;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.ICCServiceContext;
@@ -60,6 +60,10 @@
import org.apache.hyracks.hdfs.dataflow.InputSplitsFactory;
import org.apache.hyracks.hdfs.scheduler.Scheduler;
+import static org.apache.asterix.external.util.ExternalDataConstants.CONTAINER_NAME_FIELD_NAME;
+import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_PARQUET;
+import static org.apache.hyracks.api.util.ExceptionUtils.getMessageOrToString;
+
public class HDFSDataSourceFactory implements IRecordReaderFactory<Object>, IExternalDataSourceFactory {
private static final long serialVersionUID = 1L;
@@ -81,19 +85,21 @@
private InputSplit[] inputSplits;
private String nodeName;
private Class recordReaderClazz;
+ private IExternalFilterEvaluatorFactory filterEvaluatorFactory;
@Override
public void configure(IServiceContext serviceCtx, Map<String, String> configuration,
IWarningCollector warningCollector, IExternalFilterEvaluatorFactory filterEvaluatorFactory)
throws AlgebricksException, HyracksDataException {
- JobConf hdfsConf = createHdfsConf(serviceCtx, configuration);
+ JobConf hdfsConf = prepareHDFSConf(serviceCtx, configuration, filterEvaluatorFactory);
configureHdfsConf(hdfsConf, configuration);
}
- protected JobConf createHdfsConf(IServiceContext serviceCtx, Map<String, String> configuration)
- throws HyracksDataException {
+ protected JobConf prepareHDFSConf(IServiceContext serviceCtx, Map<String, String> configuration,
+ IExternalFilterEvaluatorFactory filterEvaluatorFactory) throws HyracksDataException {
this.serviceCtx = serviceCtx;
this.configuration = configuration;
+ this.filterEvaluatorFactory = filterEvaluatorFactory;
init((ICCServiceContext) serviceCtx);
return HDFSUtils.configureHDFSJobConf(configuration);
}
@@ -210,9 +216,10 @@
* 2. Indexing Stream Record Reader: When we transform the input into a byte stream and perform indexing.
* 3. HDFS Record Reader: When we simply pass the Writable object as it is to the parser.
*/
+ @SuppressWarnings("unchecked")
@Override
- public IRecordReader<? extends Object> createRecordReader(IHyracksTaskContext ctx, int partition)
- throws HyracksDataException {
+ public IRecordReader<?> createRecordReader(IExternalDataRuntimeContext context) throws HyracksDataException {
+ IHyracksTaskContext ctx = context.getTaskContext();
try {
if (recordReaderClazz != null) {
StreamRecordReader streamReader = (StreamRecordReader) recordReaderClazz.getConstructor().newInstance();
@@ -221,9 +228,8 @@
}
restoreConfig(ctx);
JobConf readerConf = conf;
- if (ctx.getWarningCollector().shouldWarn()
- && configuration.get(ExternalDataConstants.KEY_INPUT_FORMAT.trim())
- .equals(ExternalDataConstants.INPUT_FORMAT_PARQUET)) {
+ if (ctx.getWarningCollector().shouldWarn() && configuration.get(
+ ExternalDataConstants.KEY_INPUT_FORMAT.trim()).equals(ExternalDataConstants.INPUT_FORMAT_PARQUET)) {
/*
* JobConf is used to pass warnings from the ParquetReadSupport to ParquetReader. As multiple
* partitions can issue different warnings, we might have a race condition on JobConf. Thus, we
@@ -231,8 +237,7 @@
*/
readerConf = confFactory.getConf();
}
- return createRecordReader(configuration, read, inputSplits, readSchedule, nodeName, readerConf,
- ctx.getWarningCollector());
+ return createRecordReader(configuration, read, inputSplits, readSchedule, nodeName, readerConf, context);
} catch (Exception e) {
throw HyracksDataException.create(e);
}
@@ -248,12 +253,19 @@
return recordReaderNames;
}
- private static IRecordReader<? extends Object> createRecordReader(Map<String, String> configuration, boolean[] read,
+ @Override
+ public IExternalDataRuntimeContext createExternalDataRuntimeContext(IHyracksTaskContext context, int partition) {
+ IExternalFilterValueEmbedder valueEmbedder =
+ filterEvaluatorFactory.createValueEmbedder(context.getWarningCollector());
+ return new ExternalStreamRuntimeDataContext(context, partition, valueEmbedder);
+ }
+
+ private static IRecordReader<?> createRecordReader(Map<String, String> configuration, boolean[] read,
InputSplit[] inputSplits, String[] readSchedule, String nodeName, JobConf conf,
- IWarningCollector warningCollector) {
+ IExternalDataRuntimeContext context) {
if (configuration.get(ExternalDataConstants.KEY_INPUT_FORMAT.trim())
.equals(ExternalDataConstants.INPUT_FORMAT_PARQUET)) {
- return new ParquetFileRecordReader<>(read, inputSplits, readSchedule, nodeName, conf, warningCollector);
+ return new ParquetFileRecordReader<>(read, inputSplits, readSchedule, nodeName, conf, context);
} else {
return new HDFSRecordReader<>(read, inputSplits, readSchedule, nodeName, conf);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVToRecordWithMetadataAndPKConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVToRecordWithMetadataAndPKConverter.java
index 8b930aa..ad59f75 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVToRecordWithMetadataAndPKConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVToRecordWithMetadataAndPKConverter.java
@@ -23,6 +23,7 @@
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRawRecord;
import org.apache.asterix.external.input.record.CharArrayRecord;
import org.apache.asterix.external.input.record.RecordWithMetadataAndPK;
@@ -39,17 +40,19 @@
private final int valueIndex;
private final RecordWithMetadataAndPK<char[]> recordWithMetadata;
private final CharArrayRecord record;
- private LongSupplier lineNumber = ExternalDataConstants.NO_LINES;
+ private final LongSupplier lineNumber;
public CSVToRecordWithMetadataAndPKConverter(final int valueIndex, final char delimiter, final ARecordType metaType,
final ARecordType recordType, final int[] keyIndicator, final int[] keyIndexes, final IAType[] keyTypes,
- IWarningCollector warningCollector) {
+ IExternalDataRuntimeContext context) {
+ IWarningCollector warningCollector = context.getTaskContext().getWarningCollector();
this.cursor = new FieldCursorForDelimitedDataParser(null, delimiter, ExternalDataConstants.QUOTE,
warningCollector, ExternalDataConstants.EMPTY_STRING);
this.record = new CharArrayRecord();
this.valueIndex = valueIndex;
this.recordWithMetadata = new RecordWithMetadataAndPK<>(record, metaType.getFieldTypes(), recordType,
keyIndicator, keyIndexes, keyTypes);
+ lineNumber = context.getLineNumberSupplier();
}
@Override
@@ -79,9 +82,4 @@
}
return recordWithMetadata;
}
-
- @Override
- public void configure(LongSupplier lineNumber) {
- this.lineNumber = lineNumber == null ? ExternalDataConstants.NO_LINES : lineNumber;
- }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVWithRecordConverterFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVWithRecordConverterFactory.java
index a9e5bc7..1961500 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVWithRecordConverterFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/CSVWithRecordConverterFactory.java
@@ -22,12 +22,12 @@
import java.util.Map;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordConverter;
import org.apache.asterix.external.input.record.RecordWithMetadataAndPK;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.IAType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
public class CSVWithRecordConverterFactory implements IRecordConverterFactory<char[], RecordWithMetadataAndPK<char[]>> {
@@ -41,9 +41,10 @@
private IAType[] keyTypes;
@Override
- public IRecordConverter<char[], RecordWithMetadataAndPK<char[]>> createConverter(IHyracksTaskContext ctx) {
+ public IRecordConverter<char[], RecordWithMetadataAndPK<char[]>> createConverter(
+ IExternalDataRuntimeContext context) {
return new CSVToRecordWithMetadataAndPKConverter(recordIndex, delimiter, metaType, recordType, keyIndicators,
- keyIndexes, keyTypes, ctx.getWarningCollector());
+ keyIndexes, keyTypes, context);
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/DCPConverterFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/DCPConverterFactory.java
index b228b94..eb4c8e1 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/DCPConverterFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/DCPConverterFactory.java
@@ -21,10 +21,10 @@
import java.util.Map;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordConverter;
import org.apache.asterix.external.input.record.RecordWithMetadataAndPK;
import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import com.couchbase.client.core.message.dcp.DCPRequest;
@@ -51,7 +51,8 @@
}
@Override
- public IRecordConverter<DCPRequest, RecordWithMetadataAndPK<char[]>> createConverter(IHyracksTaskContext ctx) {
+ public IRecordConverter<DCPRequest, RecordWithMetadataAndPK<char[]>> createConverter(
+ IExternalDataRuntimeContext context) {
return new DCPMessageToRecordConverter();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/IRecordConverterFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/IRecordConverterFactory.java
index 875a331..96ab8ba 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/IRecordConverterFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/converter/IRecordConverterFactory.java
@@ -22,22 +22,22 @@
import java.util.Map;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordConverter;
import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
public interface IRecordConverterFactory<I, O> extends Serializable {
- public IRecordConverter<I, O> createConverter(IHyracksTaskContext ctx);
+ IRecordConverter<I, O> createConverter(IExternalDataRuntimeContext context);
- public void configure(Map<String, String> configuration) throws AsterixException;
+ void configure(Map<String, String> configuration) throws AsterixException;
- public Class<?> getInputClass();
+ Class<?> getInputClass();
- public Class<?> getOutputClass();
+ Class<?> getOutputClass();
- public void setRecordType(ARecordType recordType);
+ void setRecordType(ARecordType recordType);
- public void setMetaType(ARecordType metaType);
+ void setMetaType(ARecordType metaType);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/abstracts/AbstractExternalInputStreamFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/abstracts/AbstractExternalInputStreamFactory.java
index 34e3d2d..713a0d5 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/abstracts/AbstractExternalInputStreamFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/abstracts/AbstractExternalInputStreamFactory.java
@@ -28,8 +28,10 @@
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
import org.apache.asterix.external.api.AsterixInputStream;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IInputStreamFactory;
import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.asterix.external.provider.context.ExternalStreamRuntimeDataContext;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
@@ -74,8 +76,10 @@
}
@Override
- public IExternalFilterValueEmbedder createFilterValueEmbedder(IWarningCollector warningCollector) {
- return filterEvaluatorFactory.createValueEmbedder(warningCollector);
+ public IExternalDataRuntimeContext createExternalDataRuntimeContext(IHyracksTaskContext context, int partition) {
+ IExternalFilterValueEmbedder valueEmbedder =
+ filterEvaluatorFactory.createValueEmbedder(context.getWarningCollector());
+ return new ExternalStreamRuntimeDataContext(context, partition, valueEmbedder);
}
public static class PartitionWorkLoadBasedOnSize implements Serializable {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
index 98b5b06..f335538 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
@@ -83,7 +83,7 @@
//Configure Hadoop S3 input splits
try {
- JobConf conf = createHdfsConf(serviceCtx, configuration);
+ JobConf conf = prepareHDFSConf(serviceCtx, configuration, filterEvaluatorFactory);
int numberOfPartitions = getPartitionConstraint().getLocations().length;
configureAwsS3HdfsJobConf(conf, configuration, numberOfPartitions);
configureHdfsConf(conf, configuration);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
index 25ac181..530ce74 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
@@ -79,7 +79,7 @@
putAzureBlobConfToHadoopConf(configuration, path);
// configure Hadoop Azure input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration);
+ JobConf conf = prepareHDFSConf(serviceCtx, configuration, filterEvaluatorFactory);
configureAzureHdfsJobConf(conf, configuration, endPoint);
configureHdfsConf(conf, configuration);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureDataLakeParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureDataLakeParquetReaderFactory.java
index 3ef9c56..4dedb08 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureDataLakeParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureDataLakeParquetReaderFactory.java
@@ -66,7 +66,7 @@
putAzureDataLakeConfToHadoopConf(configuration, path);
//Configure Hadoop Azure input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration);
+ JobConf conf = prepareHDFSConf(serviceCtx, configuration, filterEvaluatorFactory);
configureAzureHdfsJobConf(conf, configuration, endPoint);
configureHdfsConf(conf, configuration);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/gcs/parquet/GCSParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/gcs/parquet/GCSParquetReaderFactory.java
index 50a1039..17cad3e 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/gcs/parquet/GCSParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/gcs/parquet/GCSParquetReaderFactory.java
@@ -69,7 +69,7 @@
putGCSConfToHadoopConf(configuration, path);
// configure hadoop input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration);
+ JobConf conf = prepareHDFSConf(serviceCtx, configuration, filterEvaluatorFactory);
int numberOfPartitions = getPartitionConstraint().getLocations().length;
GCSUtils.configureHdfsJobConf(conf, configuration, numberOfPartitions);
configureHdfsConf(conf, configuration);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MapredParquetInputFormat.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MapredParquetInputFormat.java
index d3ad968..7b10e09 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MapredParquetInputFormat.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MapredParquetInputFormat.java
@@ -25,22 +25,16 @@
import java.io.IOException;
import java.util.List;
-import org.apache.asterix.common.exceptions.ErrorCode;
-import org.apache.asterix.common.exceptions.RuntimeDataException;
-import org.apache.hadoop.mapred.FileSplit;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.data.std.api.IValueReference;
import org.apache.hyracks.data.std.primitive.VoidPointable;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
-import org.apache.hyracks.util.LogRedactionUtil;
import org.apache.parquet.hadoop.Footer;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.ParquetInputSplit;
-import org.apache.parquet.hadoop.ParquetRecordReader;
/**
* For the Original implementation, see {@code DeprecatedParquetInputFormat}
@@ -53,12 +47,13 @@
*/
public class MapredParquetInputFormat extends org.apache.hadoop.mapred.FileInputFormat<Void, VoidPointable> {
- protected ParquetInputFormat<ArrayBackedValueStorage> realInputFormat = new ParquetInputFormat<>();
+ private final ParquetInputFormat<ArrayBackedValueStorage> realInputFormat = new ParquetInputFormat<>();
+ private IExternalFilterValueEmbedder valueEmbedder;
@Override
public RecordReader<Void, VoidPointable> getRecordReader(InputSplit split, JobConf job, Reporter reporter)
throws IOException {
- return new RecordReaderWrapper(split, job, reporter);
+ return new ParquetRecordReaderWrapper(split, job, reporter, valueEmbedder);
}
@Override
@@ -84,130 +79,15 @@
return realInputFormat.getFooters(job, asList(super.listStatus(job)));
}
- private static class RecordReaderWrapper implements RecordReader<Void, VoidPointable> {
-
- private final ParquetRecordReader<IValueReference> realReader;
- private final long splitLen; // for getPos()
-
- private final VoidPointable valueContainer;
-
- private boolean firstRecord;
- private boolean eof;
-
- public RecordReaderWrapper(InputSplit oldSplit, JobConf oldJobConf, Reporter reporter) throws IOException {
- splitLen = oldSplit.getLength();
-
- try {
- realReader = new ParquetRecordReader<>(
- ParquetInputFormat.<IValueReference> getReadSupportInstance(oldJobConf),
- ParquetInputFormat.getFilter(oldJobConf));
-
- if (oldSplit instanceof ParquetInputSplitWrapper) {
- realReader.initialize(((ParquetInputSplitWrapper) oldSplit).realSplit, oldJobConf, reporter);
- } else if (oldSplit instanceof FileSplit) {
- realReader.initialize((FileSplit) oldSplit, oldJobConf, reporter);
- } else {
- throw RuntimeDataException.create(ErrorCode.INVALID_PARQUET_FILE,
- LogRedactionUtil.userData(oldSplit.toString()), "invalid file split");
- }
- valueContainer = new VoidPointable();
- firstRecord = false;
- eof = false;
- // read once to gain access to key and value objects
- if (realReader.nextKeyValue()) {
- firstRecord = true;
- valueContainer.set(realReader.getCurrentValue());
-
- } else {
- eof = true;
- }
- } catch (InterruptedException e) {
- throw new IOException(e);
- } catch (HyracksDataException | AsterixParquetRuntimeException e) {
- throw e;
- } catch (Exception e) {
- if (e.getMessage() != null && e.getMessage().contains("not a Parquet file")) {
- throw RuntimeDataException.create(ErrorCode.INVALID_PARQUET_FILE,
- LogRedactionUtil.userData(getPath(oldSplit)), "not a Parquet file");
- }
-
- throw RuntimeDataException.create(ErrorCode.UNEXPECTED_ERROR_ENCOUNTERED,
- LogRedactionUtil.userData(e.toString()));
- }
- }
-
- private String getPath(InputSplit split) {
- if (split instanceof FileSplit) {
- return ((FileSplit) split).getPath().toString();
- } else if (split instanceof ParquetInputSplitWrapper) {
- return ((ParquetInputSplitWrapper) split).realSplit.getPath().toString();
- } else {
- return split.toString();
- }
- }
-
- @Override
- public void close() throws IOException {
- realReader.close();
- }
-
- @Override
- public Void createKey() {
- return null;
- }
-
- @Override
- public VoidPointable createValue() {
- return valueContainer;
- }
-
- @Override
- public long getPos() throws IOException {
- return (long) (splitLen * getProgress());
- }
-
- @Override
- public float getProgress() throws IOException {
- try {
- return realReader.getProgress();
- } catch (InterruptedException e) {
- throw new IOException(e);
- }
- }
-
- @Override
- public boolean next(Void key, VoidPointable value) throws IOException {
- if (eof) {
- return false;
- }
-
- if (firstRecord) { // key & value are already read.
- firstRecord = false;
- value.set(valueContainer);
- return true;
- }
-
- try {
- if (realReader.nextKeyValue()) {
- if (value != null) {
- value.set(realReader.getCurrentValue());
- }
- return true;
- }
- } catch (InterruptedException e) {
- throw new IOException(e);
- }
-
- eof = true; // strictly not required, just for consistency
- return false;
- }
+ public void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
+ this.valueEmbedder = valueEmbedder;
}
public static boolean isTaskSideMetaData(JobConf job) {
return job.getBoolean(ParquetInputFormat.TASK_SIDE_METADATA, true);
}
- private static class ParquetInputSplitWrapper implements InputSplit {
+ static class ParquetInputSplitWrapper implements InputSplit {
ParquetInputSplit realSplit;
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
index 9c1d70a..bb74cd6 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
@@ -20,6 +20,7 @@
import java.io.IOException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.input.record.ValueReferenceRecord;
import org.apache.asterix.external.input.record.reader.hdfs.AbstractHDFSRecordReader;
import org.apache.asterix.external.util.HDFSUtils;
@@ -38,9 +39,10 @@
private final IWarningCollector warningCollector;
public ParquetFileRecordReader(boolean[] read, InputSplit[] inputSplits, String[] readSchedule, String nodeName,
- JobConf conf, IWarningCollector warningCollector) {
+ JobConf conf, IExternalDataRuntimeContext context) {
super(read, inputSplits, readSchedule, nodeName, new ValueReferenceRecord<>(), conf);
- this.warningCollector = warningCollector;
+ this.warningCollector = context.getTaskContext().getWarningCollector();
+ ((MapredParquetInputFormat) inputFormat).setValueEmbedder(context.getValueEmbedder());
}
@Override
@@ -59,7 +61,9 @@
@Override
protected RecordReader<Void, V> getRecordReader(int splitIndex) throws IOException {
try {
- reader = (RecordReader<Void, V>) inputFormat.getRecordReader(inputSplits[splitIndex], conf, Reporter.NULL);
+ ParquetRecordReaderWrapper readerWrapper = (ParquetRecordReaderWrapper) inputFormat
+ .getRecordReader(inputSplits[splitIndex], conf, Reporter.NULL);
+ reader = (RecordReader<Void, V>) readerWrapper;
} catch (AsterixParquetRuntimeException e) {
throw e.getHyracksDataException();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
index 797a2b2..f88ee80 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
@@ -23,6 +23,7 @@
import java.util.List;
import java.util.Map;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.input.record.reader.hdfs.parquet.converter.ParquetConverterContext;
import org.apache.asterix.external.input.record.reader.hdfs.parquet.converter.nested.RootConverter;
import org.apache.asterix.external.util.HDFSUtils;
@@ -38,6 +39,8 @@
import org.apache.parquet.schema.MessageType;
public class ParquetReadSupport extends ReadSupport<IValueReference> {
+ private IExternalFilterValueEmbedder valueEmbedder;
+
@Override
public ReadContext init(InitContext context) {
MessageType requestedSchema = getRequestedSchema(context);
@@ -47,7 +50,7 @@
@Override
public RecordMaterializer<IValueReference> prepareForRead(Configuration configuration,
Map<String, String> keyValueMetaData, MessageType fileSchema, ReadContext readContext) {
- return new ADMRecordMaterializer(configuration, readContext);
+ return new ADMRecordMaterializer(configuration, valueEmbedder, readContext);
}
private static MessageType getRequestedSchema(InitContext initContext) {
@@ -75,14 +78,19 @@
}
}
+ void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
+ this.valueEmbedder = valueEmbedder;
+ }
+
private static class ADMRecordMaterializer extends RecordMaterializer<IValueReference> {
private final RootConverter rootConverter;
private final List<Warning> warnings;
private final Configuration configuration;
- public ADMRecordMaterializer(Configuration configuration, ReadContext readContext) {
+ public ADMRecordMaterializer(Configuration configuration, IExternalFilterValueEmbedder valueEmbedder,
+ ReadContext readContext) {
warnings = new ArrayList<>();
- rootConverter = new RootConverter(readContext.getRequestedSchema(), configuration, warnings);
+ rootConverter = new RootConverter(readContext.getRequestedSchema(), valueEmbedder, configuration, warnings);
this.configuration = configuration;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetRecordReaderWrapper.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetRecordReaderWrapper.java
new file mode 100644
index 0000000..fe95d28
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetRecordReaderWrapper.java
@@ -0,0 +1,161 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet;
+
+import java.io.IOException;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.util.LogRedactionUtil;
+import org.apache.parquet.hadoop.ParquetInputFormat;
+import org.apache.parquet.hadoop.ParquetRecordReader;
+import org.apache.parquet.hadoop.api.ReadSupport;
+
+public class ParquetRecordReaderWrapper implements RecordReader<Void, VoidPointable> {
+ private final ParquetRecordReader<IValueReference> realReader;
+ private final long splitLen; // for getPos()
+
+ private final VoidPointable valueContainer;
+
+ private boolean firstRecord;
+ private boolean eof;
+
+ public ParquetRecordReaderWrapper(InputSplit oldSplit, JobConf oldJobConf, Reporter reporter,
+ IExternalFilterValueEmbedder valueEmbedder) throws IOException {
+ splitLen = oldSplit.getLength();
+
+ try {
+ ReadSupport<IValueReference> readSupport = ParquetInputFormat.getReadSupportInstance(oldJobConf);
+ ParquetReadSupport parquetReadSupport = (ParquetReadSupport) readSupport;
+ parquetReadSupport.setValueEmbedder(valueEmbedder);
+ realReader = new ParquetRecordReader<>(readSupport, ParquetInputFormat.getFilter(oldJobConf));
+
+ if (oldSplit instanceof MapredParquetInputFormat.ParquetInputSplitWrapper) {
+ realReader.initialize(((MapredParquetInputFormat.ParquetInputSplitWrapper) oldSplit).realSplit,
+ oldJobConf, reporter);
+ } else if (oldSplit instanceof FileSplit) {
+ realReader.initialize((FileSplit) oldSplit, oldJobConf, reporter);
+ } else {
+ throw RuntimeDataException.create(ErrorCode.INVALID_PARQUET_FILE,
+ LogRedactionUtil.userData(oldSplit.toString()), "invalid file split");
+ }
+
+ // Set the path for value embedder
+ valueEmbedder.setPath(getPath(oldSplit));
+
+ valueContainer = new VoidPointable();
+ firstRecord = false;
+ eof = false;
+ // read once to gain access to key and value objects
+ if (realReader.nextKeyValue()) {
+ firstRecord = true;
+ valueContainer.set(realReader.getCurrentValue());
+ } else {
+ eof = true;
+ }
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ } catch (HyracksDataException | AsterixParquetRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ if (e.getMessage() != null && e.getMessage().contains("not a Parquet file")) {
+ throw RuntimeDataException.create(ErrorCode.INVALID_PARQUET_FILE,
+ LogRedactionUtil.userData(getPath(oldSplit)), "not a Parquet file");
+ }
+
+ throw RuntimeDataException.create(ErrorCode.UNEXPECTED_ERROR_ENCOUNTERED,
+ LogRedactionUtil.userData(e.toString()));
+ }
+ }
+
+ private String getPath(InputSplit split) {
+ if (split instanceof FileSplit) {
+ return ((FileSplit) split).getPath().toString();
+ } else if (split instanceof MapredParquetInputFormat.ParquetInputSplitWrapper) {
+ return ((MapredParquetInputFormat.ParquetInputSplitWrapper) split).realSplit.getPath().toString();
+ } else {
+ return split.toString();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ realReader.close();
+ }
+
+ @Override
+ public Void createKey() {
+ return null;
+ }
+
+ @Override
+ public VoidPointable createValue() {
+ return valueContainer;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return (long) (splitLen * getProgress());
+ }
+
+ @Override
+ public float getProgress() throws IOException {
+ try {
+ return realReader.getProgress();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public boolean next(Void key, VoidPointable value) throws IOException {
+ if (eof) {
+ return false;
+ }
+
+ if (firstRecord) { // key & value are already read.
+ firstRecord = false;
+ value.set(valueContainer);
+ return true;
+ }
+
+ try {
+ if (realReader.nextKeyValue()) {
+ if (value != null) {
+ value.set(realReader.getCurrentValue());
+ }
+ return true;
+ }
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+
+ eof = true; // strictly not required, just for consistency
+ return false;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java
index 4982ca5..29e1a14 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java
@@ -25,6 +25,8 @@
import org.apache.asterix.dataflow.data.nontagged.serde.ABinarySerializerDeserializer;
import org.apache.asterix.dataflow.data.nontagged.serde.AStringSerializerDeserializer;
+import org.apache.asterix.external.input.filter.NoOpFilterValueEmbedder;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.input.stream.StandardUTF8ToModifiedUTF8DataOutput;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.external.util.ExternalDataConstants.ParquetOptions;
@@ -119,8 +121,21 @@
private final String timeZoneId;
private final int timeZoneOffset;
+ /*
+ * ************************************************************************
+ * Value Embedder
+ * ************************************************************************
+ */
+ private final IExternalFilterValueEmbedder valueEmbedder;
+
public ParquetConverterContext(Configuration configuration, List<Warning> warnings) {
+ this(configuration, NoOpFilterValueEmbedder.INSTANCE, warnings);
+ }
+
+ public ParquetConverterContext(Configuration configuration, IExternalFilterValueEmbedder valueEmbedder,
+ List<Warning> warnings) {
this.warnings = warnings;
+ this.valueEmbedder = valueEmbedder;
modifiedUTF8DataOutput = new StandardUTF8ToModifiedUTF8DataOutput(
new AStringSerializerDeserializer(new UTF8StringWriter(), new UTF8StringReader()));
@@ -137,6 +152,10 @@
}
}
+ public IExternalFilterValueEmbedder getValueEmbedder() {
+ return valueEmbedder;
+ }
+
public List<Warning> getWarnings() {
return warnings;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java
index 24a531a..89b620c 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java
@@ -21,6 +21,7 @@
import java.io.DataOutput;
import java.util.List;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.input.record.reader.hdfs.parquet.converter.ParquetConverterContext;
import org.apache.hadoop.conf.Configuration;
import org.apache.hyracks.api.exceptions.Warning;
@@ -31,8 +32,9 @@
public class RootConverter extends ObjectConverter {
private final ArrayBackedValueStorage rootBuffer;
- public RootConverter(GroupType parquetType, Configuration configuration, List<Warning> warnings) {
- super(null, -1, parquetType, new ParquetConverterContext(configuration, warnings));
+ public RootConverter(GroupType parquetType, IExternalFilterValueEmbedder valueEmbedder, Configuration configuration,
+ List<Warning> warnings) {
+ super(null, -1, parquetType, new ParquetConverterContext(configuration, valueEmbedder, warnings));
this.rootBuffer = new ArrayBackedValueStorage();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/http/HttpServerRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/http/HttpServerRecordReaderFactory.java
index b5ade31..4736675 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/http/HttpServerRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/http/HttpServerRecordReaderFactory.java
@@ -24,6 +24,7 @@
import org.apache.asterix.common.exceptions.CompilationException;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
import org.apache.asterix.external.util.ExternalDataConstants;
@@ -32,7 +33,6 @@
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.IWarningCollector;
import org.apache.hyracks.http.server.HttpServerConfigBuilder;
@@ -54,10 +54,10 @@
private List<Pair<String, Integer>> serverAddrs;
@Override
- public IRecordReader<? extends char[]> createRecordReader(IHyracksTaskContext ctx, int partition)
+ public IRecordReader<? extends char[]> createRecordReader(IExternalDataRuntimeContext context)
throws HyracksDataException {
try {
- return new HttpServerRecordReader(serverAddrs.get(partition).getRight(), entryPoint, queueSize,
+ return new HttpServerRecordReader(serverAddrs.get(context.getPartition()).getRight(), entryPoint, queueSize,
HttpServerConfigBuilder.createDefault());
} catch (Exception e) {
throw HyracksDataException.create(e);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
index 0026f07..21ab53e 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
@@ -27,6 +27,7 @@
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
@@ -34,7 +35,6 @@
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.IWarningCollector;
@@ -86,10 +86,10 @@
}
@Override
- public IRecordReader<? extends SyndEntry> createRecordReader(IHyracksTaskContext ctx, int partition)
+ public IRecordReader<? extends SyndEntry> createRecordReader(IExternalDataRuntimeContext context)
throws HyracksDataException {
try {
- return new RSSRecordReader(urls.get(partition));
+ return new RSSRecordReader(urls.get(context.getPartition()));
} catch (MalformedURLException e) {
throw HyracksDataException.create(e);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
index 9d12325..937f3fe 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
@@ -31,7 +31,6 @@
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IStreamNotificationHandler;
import org.apache.asterix.external.dataflow.AbstractFeedDataFlowController;
-import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.input.record.CharArrayRecord;
import org.apache.asterix.external.input.stream.AsterixInputStreamReader;
import org.apache.asterix.external.util.ExternalDataUtils;
@@ -49,11 +48,9 @@
protected IFeedLogManager feedLogManager;
private Supplier<String> dataSourceName = EMPTY_STRING;
private Supplier<String> previousDataSourceName = EMPTY_STRING;
- private AsterixInputStream inputStream;
public void configure(AsterixInputStream inputStream, Map<String, String> config) {
int bufferSize = ExternalDataUtils.getOrDefaultBufferSize(config);
- this.inputStream = inputStream;
this.reader = new AsterixInputStreamReader(inputStream, bufferSize);
record = new CharArrayRecord();
inputBuffer = new char[bufferSize];
@@ -123,11 +120,6 @@
return dataSourceName;
}
- @Override
- public void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
- inputStream.setValueEmbedder(valueEmbedder);
- }
-
String getPreviousStreamName() {
return previousDataSourceName.get();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReaderFactory.java
index c9567c5..c736961 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReaderFactory.java
@@ -27,6 +27,7 @@
import org.apache.asterix.common.exceptions.CompilationException;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IInputStreamFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
@@ -35,6 +36,7 @@
import org.apache.asterix.external.input.stream.factory.SocketClientInputStreamFactory;
import org.apache.asterix.external.input.stream.factory.SocketServerInputStreamFactory;
import org.apache.asterix.external.provider.StreamRecordReaderProvider;
+import org.apache.asterix.external.provider.context.ExternalReaderRuntimeDataContext;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -50,8 +52,8 @@
protected Map<String, String> configuration;
protected Class recordReaderClazz;
protected IExternalFilterEvaluatorFactory filterEvaluatorFactory;
- private static final List<String> recordReaderNames =
- Collections.unmodifiableList(Arrays.asList(ExternalDataConstants.ALIAS_LOCALFS_ADAPTER,
+ private static final List<String> recordReaderNames = Collections.unmodifiableList(
+ Arrays.asList(ExternalDataConstants.ALIAS_LOCALFS_ADAPTER,
ExternalDataConstants.KEY_ALIAS_ADAPTER_NAME_SOCKET, ExternalDataConstants.KEY_ADAPTER_NAME_SOCKET,
ExternalDataConstants.STREAM_SOCKET_CLIENT));
@@ -82,22 +84,19 @@
}
@Override
- public final IRecordReader<? extends char[]> createRecordReader(IHyracksTaskContext ctx, int partition)
+ public final IRecordReader<? extends char[]> createRecordReader(IExternalDataRuntimeContext context)
throws HyracksDataException {
- try {
- StreamRecordReader streamRecordReader =
- (StreamRecordReader) recordReaderClazz.getConstructor().newInstance();
- streamRecordReader.configure(ctx, streamFactory.createInputStream(ctx, partition), configuration);
- return streamRecordReader;
- } catch (InstantiationException | IllegalAccessException | InvocationTargetException
- | NoSuchMethodException e) {
- throw HyracksDataException.create(e);
- }
+ StreamRecordReader reader = createReader(context.getTaskContext(), context.getPartition());
+ ((ExternalReaderRuntimeDataContext) context).setReader(reader);
+ return reader;
}
@Override
- public final IExternalFilterValueEmbedder createFilterValueEmbedder(IWarningCollector warningCollector) {
- return filterEvaluatorFactory.createValueEmbedder(warningCollector);
+ public final IExternalDataRuntimeContext createExternalDataRuntimeContext(IHyracksTaskContext context,
+ int partition) throws HyracksDataException {
+ IExternalFilterValueEmbedder valueEmbedder =
+ filterEvaluatorFactory.createValueEmbedder(context.getWarningCollector());
+ return new ExternalReaderRuntimeDataContext(context, partition, valueEmbedder);
}
@Override
@@ -109,8 +108,8 @@
String reader = config.get(ExternalDataConstants.KEY_READER);
if (reader.equals(ExternalDataConstants.ALIAS_LOCALFS_ADAPTER)) {
streamFactory = new LocalFSInputStreamFactory();
- } else if (reader.equals(ExternalDataConstants.KEY_ALIAS_ADAPTER_NAME_SOCKET)
- || reader.equals(ExternalDataConstants.KEY_ADAPTER_NAME_SOCKET)) {
+ } else if (reader.equals(ExternalDataConstants.KEY_ALIAS_ADAPTER_NAME_SOCKET) || reader.equals(
+ ExternalDataConstants.KEY_ADAPTER_NAME_SOCKET)) {
streamFactory = new SocketServerInputStreamFactory();
} else if (reader.equals(ExternalDataConstants.STREAM_SOCKET_CLIENT)) {
streamFactory = new SocketClientInputStreamFactory();
@@ -118,4 +117,17 @@
throw new CompilationException(ErrorCode.FEED_UNKNOWN_ADAPTER_NAME);
}
}
+
+ @SuppressWarnings("unchecked")
+ private StreamRecordReader createReader(IHyracksTaskContext ctx, int partition) throws HyracksDataException {
+ try {
+ StreamRecordReader streamRecordReader =
+ (StreamRecordReader) recordReaderClazz.getConstructor().newInstance();
+ streamRecordReader.configure(ctx, streamFactory.createInputStream(ctx, partition), configuration);
+ return streamRecordReader;
+ } catch (InstantiationException | IllegalAccessException | InvocationTargetException |
+ NoSuchMethodException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
index ef55d4a..2f87074 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
@@ -27,6 +27,7 @@
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
@@ -37,7 +38,6 @@
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.IWarningCollector;
import org.apache.logging.log4j.LogManager;
@@ -127,7 +127,7 @@
}
@Override
- public IRecordReader<? extends char[]> createRecordReader(IHyracksTaskContext ctx, int partition)
+ public IRecordReader<? extends char[]> createRecordReader(IExternalDataRuntimeContext context)
throws HyracksDataException {
IRecordReader<? extends char[]> recordReader;
switch (configuration.get(ExternalDataConstants.KEY_READER)) {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/AbstractJsonDataParser.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/AbstractJsonDataParser.java
index d9e374f..a5ac07f 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/AbstractJsonDataParser.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/AbstractJsonDataParser.java
@@ -30,12 +30,11 @@
import org.apache.asterix.builders.IAsterixListBuilder;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
-import org.apache.asterix.external.input.filter.NoOpFilterValueEmbedder;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.parser.jackson.ADMToken;
import org.apache.asterix.external.parser.jackson.GeometryCoParser;
import org.apache.asterix.external.parser.jackson.ParserContext;
-import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.om.base.ABoolean;
import org.apache.asterix.om.base.ANull;
import org.apache.asterix.om.base.AUnorderedList;
@@ -69,27 +68,29 @@
protected final JsonFactory jsonFactory;
protected final ARecordType rootType;
protected final GeometryCoParser geometryCoParser;
- protected Supplier<String> dataSourceName = ExternalDataConstants.EMPTY_STRING;
- protected LongSupplier lineNumber = ExternalDataConstants.NO_LINES;
+ protected final Supplier<String> dataSourceName;
+ protected final LongSupplier lineNumber;
+ protected final IExternalFilterValueEmbedder valueEmbedder;
protected JsonParser jsonParser;
- protected IExternalFilterValueEmbedder valueEmbedder;
-
/**
* Initialize JSONDataParser with GeometryCoParser
*
* @param recordType defined type.
* @param jsonFactory Jackson JSON parser factory.
*/
- public AbstractJsonDataParser(ARecordType recordType, JsonFactory jsonFactory) {
+ public AbstractJsonDataParser(ARecordType recordType, JsonFactory jsonFactory,
+ IExternalDataRuntimeContext context) {
// recordType currently cannot be null, however this is to guarantee for any future changes.
this.rootType = recordType != null ? recordType : RecordUtil.FULLY_OPEN_RECORD_TYPE;
this.jsonFactory = jsonFactory;
//GeometryCoParser to parse GeoJSON objects to AsterixDB internal spatial types.
geometryCoParser = new GeometryCoParser(jsonParser);
parserContext = new ParserContext();
- valueEmbedder = NoOpFilterValueEmbedder.INSTANCE;
+ dataSourceName = context.getDatasourceNameSupplier();
+ lineNumber = context.getLineNumberSupplier();
+ valueEmbedder = context.getValueEmbedder();
}
/*
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/DelimitedDataParser.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/DelimitedDataParser.java
index 2936d11..bc855b4 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/DelimitedDataParser.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/DelimitedDataParser.java
@@ -34,17 +34,16 @@
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
import org.apache.asterix.external.api.IDataParser;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRawRecord;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IStreamDataParser;
-import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.om.base.AMutableString;
import org.apache.asterix.om.typecomputer.impl.TypeComputeUtils;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.IAType;
import org.apache.asterix.om.utils.NonTaggedFormatUtil;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.IWarningCollector;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
@@ -65,20 +64,20 @@
private final ArrayBackedValueStorage fieldValueBuffer;
private final DataOutput fieldValueBufferOutput;
private final IValueParser[] valueParsers;
- private FieldCursorForDelimitedDataParser cursor;
- private Supplier<String> dataSourceName;
- private LongSupplier lineNumber;
+ private final Supplier<String> dataSourceName;
+ private final LongSupplier lineNumber;
private final byte[] fieldTypeTags;
private final int[] fldIds;
private final ArrayBackedValueStorage[] nameBuffers;
private final char[] nullChars;
+ private FieldCursorForDelimitedDataParser cursor;
- public DelimitedDataParser(IHyracksTaskContext ctx, IValueParserFactory[] valueParserFactories, char fieldDelimiter,
- char quote, boolean hasHeader, ARecordType recordType, boolean isStreamParser, String nullString)
- throws HyracksDataException {
- this.dataSourceName = ExternalDataConstants.EMPTY_STRING;
- this.lineNumber = ExternalDataConstants.NO_LINES;
- this.warnings = ctx.getWarningCollector();
+ public DelimitedDataParser(IExternalDataRuntimeContext context, IValueParserFactory[] valueParserFactories,
+ char fieldDelimiter, char quote, boolean hasHeader, ARecordType recordType, boolean isStreamParser,
+ String nullString) throws HyracksDataException {
+ this.dataSourceName = context.getDatasourceNameSupplier();
+ this.lineNumber = context.getLineNumberSupplier();
+ this.warnings = context.getTaskContext().getWarningCollector();
this.fieldDelimiter = fieldDelimiter;
this.quote = quote;
this.hasHeader = hasHeader;
@@ -245,13 +244,6 @@
return true;
}
- @Override
- public void configure(Supplier<String> dataSourceName, LongSupplier lineNumber) {
- this.dataSourceName = dataSourceName == null ? ExternalDataConstants.EMPTY_STRING : dataSourceName;
- this.lineNumber = lineNumber == null ? ExternalDataConstants.NO_LINES : lineNumber;
-
- }
-
private String getDataSourceName() {
return dataSourceName.get();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/JSONDataParser.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/JSONDataParser.java
index 20b466b..3f1d86b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/JSONDataParser.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/JSONDataParser.java
@@ -18,43 +18,51 @@
*/
package org.apache.asterix.external.parser;
-import static org.apache.asterix.common.exceptions.ErrorCode.PARSER_DATA_PARSER_UNEXPECTED_TOKEN;
-
-import java.io.DataOutput;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.function.LongSupplier;
-import java.util.function.Supplier;
-
-import org.apache.asterix.common.exceptions.ErrorCode;
-import org.apache.asterix.common.exceptions.RuntimeDataException;
-import org.apache.asterix.external.api.IRawRecord;
-import org.apache.asterix.external.api.IRecordDataParser;
-import org.apache.asterix.external.api.IStreamDataParser;
-import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
-import org.apache.asterix.external.parser.jackson.ADMToken;
-import org.apache.asterix.external.util.ExternalDataConstants;
-import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.TreeTraversingParser;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
+import org.apache.asterix.external.api.IRawRecord;
+import org.apache.asterix.external.api.IRecordDataParser;
+import org.apache.asterix.external.api.IStreamDataParser;
+import org.apache.asterix.external.parser.jackson.ADMToken;
+import org.apache.asterix.external.provider.context.NoOpExternalRuntimeDataContext;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+import static org.apache.asterix.common.exceptions.ErrorCode.PARSER_DATA_PARSER_UNEXPECTED_TOKEN;
+
/**
* JSON format parser using Jackson parser.
*/
public class JSONDataParser extends AbstractJsonDataParser implements IStreamDataParser, IRecordDataParser<char[]> {
/**
- * Initialize JSONDataParser
+ * This constructor is used by several query runtime evaluators and tests
*
* @param recordType defined type.
* @param jsonFactory Jackson JSON parser factory.
*/
public JSONDataParser(ARecordType recordType, JsonFactory jsonFactory) {
- super(recordType, jsonFactory);
+ this(recordType, jsonFactory, NoOpExternalRuntimeDataContext.INSTANCE);
+ }
+
+ /**
+ * Initialize JSONDataParser
+ *
+ * @param recordType defined type.
+ * @param jsonFactory Jackson JSON parser factory.
+ */
+ public JSONDataParser(ARecordType recordType, JsonFactory jsonFactory, IExternalDataRuntimeContext context) {
+ super(recordType, jsonFactory, context);
}
@Override
@@ -62,11 +70,6 @@
setInput(jsonFactory.createParser(in));
}
- @Override
- public void setValueEmbedder(IExternalFilterValueEmbedder valueEmbedder) {
- this.valueEmbedder = valueEmbedder;
- }
-
public void setInputNode(JsonNode node) {
setInput(new TreeTraversingParser(node));
}
@@ -94,12 +97,6 @@
}
@Override
- public void configure(Supplier<String> dataSourceName, LongSupplier lineNumber) {
- this.dataSourceName = dataSourceName == null ? ExternalDataConstants.EMPTY_STRING : dataSourceName;
- this.lineNumber = lineNumber == null ? ExternalDataConstants.NO_LINES : lineNumber;
- }
-
- @Override
public boolean parse(DataOutput out) throws HyracksDataException {
try {
if (nextToken() == ADMToken.EOF) {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/RecordWithMetadataParser.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/RecordWithMetadataParser.java
index 820775c..d799f22 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/RecordWithMetadataParser.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/RecordWithMetadataParser.java
@@ -20,8 +20,6 @@
import java.io.DataOutput;
import java.io.IOException;
-import java.util.function.LongSupplier;
-import java.util.function.Supplier;
import org.apache.asterix.builders.RecordBuilder;
import org.apache.asterix.external.api.IDataParser;
@@ -113,10 +111,4 @@
public void appendLastParsedPrimaryKeyToTuple(ArrayTupleBuilder tb) throws HyracksDataException {
rwm.appendPrimaryKeyToTuple(tb);
}
-
- @Override
- public void configure(Supplier<String> dataSourceName, LongSupplier lineNumber) {
- this.recordParser.configure(dataSourceName, lineNumber);
- this.converter.configure(lineNumber);
- }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/evaluators/StringJsonParseEval.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/evaluators/StringJsonParseEval.java
index 7589e33..9cb9880 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/evaluators/StringJsonParseEval.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/evaluators/StringJsonParseEval.java
@@ -18,14 +18,13 @@
*/
package org.apache.asterix.external.parser.evaluators;
-import static org.apache.asterix.om.functions.BuiltinFunctions.STRING_PARSE_JSON;
-
import java.io.DataOutput;
import java.io.IOException;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.external.parser.JSONDataParser;
import org.apache.asterix.external.parser.factory.JSONDataParserFactory;
+import org.apache.asterix.external.provider.context.DefaultExternalRuntimeDataContext;
import org.apache.asterix.om.exceptions.ExceptionUtil;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
@@ -42,6 +41,8 @@
import org.apache.hyracks.data.std.util.ByteArrayAccessibleInputStream;
import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+import static org.apache.asterix.om.functions.BuiltinFunctions.STRING_PARSE_JSON;
+
public class StringJsonParseEval implements IScalarEvaluator {
private final IEvaluatorContext ctx;
private final IScalarEvaluator inputEval;
@@ -57,7 +58,8 @@
throws IOException {
this.ctx = ctx;
this.inputEval = inputEval;
- this.parser = (JSONDataParser) new JSONDataParserFactory().createInputStreamParser(ctx.getTaskContext(), 0);
+ this.parser = (JSONDataParser) new JSONDataParserFactory().createInputStreamParser(
+ new DefaultExternalRuntimeDataContext(ctx.getTaskContext()));
this.sourceLocation = sourceLocation;
inputVal = new VoidPointable();
utf8Val = new UTF8StringPointable();
@@ -111,4 +113,4 @@
throw HyracksDataException.create(e);
}
}
-}
\ No newline at end of file
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/ADMDataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/ADMDataParserFactory.java
index 394fcb3..86af2c3 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/ADMDataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/ADMDataParserFactory.java
@@ -22,13 +22,13 @@
import java.util.Collections;
import java.util.List;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory.DataSourceType;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IStreamDataParser;
import org.apache.asterix.external.parser.ADMDataParser;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
public class ADMDataParserFactory extends AbstractRecordStreamParserFactory<char[]> {
@@ -38,7 +38,7 @@
Collections.unmodifiableList(Arrays.asList("adm", "semi-structured"));
@Override
- public IRecordDataParser<char[]> createRecordParser(IHyracksTaskContext ctx) {
+ public IRecordDataParser<char[]> createRecordParser(IExternalDataRuntimeContext context) {
return createParser();
}
@@ -53,8 +53,7 @@
}
@Override
- public IStreamDataParser createInputStreamParser(IHyracksTaskContext ctx, int partition)
- throws HyracksDataException {
+ public IStreamDataParser createInputStreamParser(IExternalDataRuntimeContext context) throws HyracksDataException {
return createParser();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/DelimitedDataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/DelimitedDataParserFactory.java
index 09f9697..8b88a69 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/DelimitedDataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/DelimitedDataParserFactory.java
@@ -22,6 +22,7 @@
import java.util.Collections;
import java.util.List;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory.DataSourceType;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IStreamDataParser;
@@ -29,7 +30,6 @@
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.data.parsers.IValueParserFactory;
@@ -41,17 +41,18 @@
ExternalDataConstants.FORMAT_DELIMITED_TEXT, ExternalDataConstants.FORMAT_TSV));
@Override
- public IRecordDataParser<char[]> createRecordParser(IHyracksTaskContext ctx) throws HyracksDataException {
- return createParser(ctx);
+ public IRecordDataParser<char[]> createRecordParser(IExternalDataRuntimeContext context)
+ throws HyracksDataException {
+ return createParser(context);
}
- private DelimitedDataParser createParser(IHyracksTaskContext ctx) throws HyracksDataException {
+ private DelimitedDataParser createParser(IExternalDataRuntimeContext context) throws HyracksDataException {
IValueParserFactory[] valueParserFactories = ExternalDataUtils.getValueParserFactories(recordType);
char delimiter = ExternalDataUtils.validateGetDelimiter(configuration);
char quote = ExternalDataUtils.validateGetQuote(configuration, delimiter);
boolean hasHeader = ExternalDataUtils.hasHeader(configuration);
String nullString = configuration.get(ExternalDataConstants.KEY_NULL_STR);
- return new DelimitedDataParser(ctx, valueParserFactories, delimiter, quote, hasHeader, recordType,
+ return new DelimitedDataParser(context, valueParserFactories, delimiter, quote, hasHeader, recordType,
ExternalDataUtils.getDataSourceType(configuration).equals(DataSourceType.STREAM), nullString);
}
@@ -61,9 +62,8 @@
}
@Override
- public IStreamDataParser createInputStreamParser(IHyracksTaskContext ctx, int partition)
- throws HyracksDataException {
- return createParser(ctx);
+ public IStreamDataParser createInputStreamParser(IExternalDataRuntimeContext context) throws HyracksDataException {
+ return createParser(context);
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/JSONDataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/JSONDataParserFactory.java
index f9d50d3..ccd5025 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/JSONDataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/JSONDataParserFactory.java
@@ -24,6 +24,7 @@
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IStreamDataParser;
import org.apache.asterix.external.parser.JSONDataParser;
@@ -33,7 +34,6 @@
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.AUnionType;
import org.apache.asterix.om.types.IAType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import com.fasterxml.jackson.core.JsonFactory;
@@ -58,9 +58,8 @@
}
@Override
- public IStreamDataParser createInputStreamParser(IHyracksTaskContext ctx, int partition)
- throws HyracksDataException {
- return createParser();
+ public IStreamDataParser createInputStreamParser(IExternalDataRuntimeContext context) throws HyracksDataException {
+ return createParser(context);
}
@Override
@@ -74,8 +73,9 @@
}
@Override
- public IRecordDataParser<char[]> createRecordParser(IHyracksTaskContext ctx) throws HyracksDataException {
- return createParser();
+ public IRecordDataParser<char[]> createRecordParser(IExternalDataRuntimeContext context)
+ throws HyracksDataException {
+ return createParser(context);
}
@Override
@@ -83,8 +83,8 @@
return char[].class;
}
- private JSONDataParser createParser() throws HyracksDataException {
- return new JSONDataParser(recordType, jsonFactory);
+ private JSONDataParser createParser(IExternalDataRuntimeContext context) throws HyracksDataException {
+ return new JSONDataParser(recordType, jsonFactory, context);
}
/*
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/NoOpDataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/NoOpDataParserFactory.java
index f4fca17..82835bb 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/NoOpDataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/NoOpDataParserFactory.java
@@ -23,13 +23,13 @@
import java.util.Map;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IRecordDataParserFactory;
import org.apache.asterix.external.parser.NoOpDataParser;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.om.types.ARecordType;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.data.std.api.IValueReference;
@@ -45,7 +45,7 @@
@Override
public void setRecordType(ARecordType recordType) throws AsterixException {
- //it always return open type
+ //it always returns open type
}
@Override
@@ -59,7 +59,8 @@
}
@Override
- public IRecordDataParser<IValueReference> createRecordParser(IHyracksTaskContext ctx) throws HyracksDataException {
+ public IRecordDataParser<IValueReference> createRecordParser(IExternalDataRuntimeContext context)
+ throws HyracksDataException {
return new NoOpDataParser();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RSSParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RSSParserFactory.java
index a1d2298..13c2ec6 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RSSParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RSSParserFactory.java
@@ -23,11 +23,11 @@
import java.util.List;
import java.util.Map;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IRecordDataParserFactory;
import org.apache.asterix.external.parser.RSSParser;
import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import com.rometools.rome.feed.synd.SyndEntry;
@@ -48,7 +48,7 @@
}
@Override
- public IRecordDataParser<SyndEntry> createRecordParser(IHyracksTaskContext ctx) {
+ public IRecordDataParser<SyndEntry> createRecordParser(IExternalDataRuntimeContext context) {
RSSParser dataParser = new RSSParser(recordType);
return dataParser;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RecordWithMetadataParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RecordWithMetadataParserFactory.java
index 3ae9b82..f9093dc 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RecordWithMetadataParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/RecordWithMetadataParserFactory.java
@@ -25,6 +25,7 @@
import org.apache.asterix.common.exceptions.CompilationException;
import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IRecordDataParserFactory;
import org.apache.asterix.external.input.record.RecordWithMetadataAndPK;
@@ -36,7 +37,6 @@
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.om.types.ARecordType;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
public class RecordWithMetadataParserFactory<I, O> implements IRecordDataParserFactory<I> {
@@ -98,9 +98,9 @@
}
@Override
- public IRecordDataParser<I> createRecordParser(IHyracksTaskContext ctx) throws HyracksDataException {
- IRecordDataParser<O> recordParser = recordParserFactory.createRecordParser(ctx);
- return new RecordWithMetadataParser<I, O>(metaType, recordParser, converterFactory.createConverter(ctx));
+ public IRecordDataParser<I> createRecordParser(IExternalDataRuntimeContext context) throws HyracksDataException {
+ IRecordDataParser<O> recordParser = recordParserFactory.createRecordParser(context);
+ return new RecordWithMetadataParser<I, O>(metaType, recordParser, converterFactory.createConverter(context));
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/TweetParserFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/TweetParserFactory.java
index e5b6d01..6083e21 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/TweetParserFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/factory/TweetParserFactory.java
@@ -23,12 +23,12 @@
import java.util.List;
import java.util.Map;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IRecordDataParserFactory;
import org.apache.asterix.external.parser.TweetParser;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.om.types.ARecordType;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
public class TweetParserFactory implements IRecordDataParserFactory<char[]> {
@@ -48,9 +48,8 @@
}
@Override
- public IRecordDataParser<char[]> createRecordParser(IHyracksTaskContext ctx) {
- TweetParser dataParser = new TweetParser(recordType);
- return dataParser;
+ public IRecordDataParser<char[]> createRecordParser(IExternalDataRuntimeContext context) {
+ return new TweetParser(recordType);
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
index 83ec7e7..574b2d1 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
@@ -27,6 +27,7 @@
import org.apache.asterix.external.api.AsterixInputStream;
import org.apache.asterix.external.api.IDataFlowController;
import org.apache.asterix.external.api.IDataParserFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IInputStreamFactory;
import org.apache.asterix.external.api.IRecordDataParser;
@@ -44,7 +45,6 @@
import org.apache.asterix.external.dataflow.FeedWithMetaDataFlowController;
import org.apache.asterix.external.dataflow.RecordDataFlowController;
import org.apache.asterix.external.dataflow.StreamDataFlowController;
-import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.external.util.IFeedLogManager;
import org.apache.asterix.om.types.ARecordType;
@@ -62,19 +62,14 @@
int partition, IExternalDataSourceFactory dataSourceFactory, IDataParserFactory dataParserFactory,
Map<String, String> configuration, boolean isFeed, IFeedLogManager feedLogManager)
throws HyracksDataException {
- IExternalFilterValueEmbedder valueEmbedder =
- dataSourceFactory.createFilterValueEmbedder(ctx.getWarningCollector());
+ IExternalDataRuntimeContext runtimeContext = dataSourceFactory.createExternalDataRuntimeContext(ctx, partition);
try {
switch (dataSourceFactory.getDataSourceType()) {
case RECORDS:
IRecordReaderFactory<?> recordReaderFactory = (IRecordReaderFactory<?>) dataSourceFactory;
- IRecordReader<?> recordReader = recordReaderFactory.createRecordReader(ctx, partition);
- recordReader.setValueEmbedder(valueEmbedder);
+ IRecordReader<?> recordReader = recordReaderFactory.createRecordReader(runtimeContext);
IRecordDataParserFactory<?> recordParserFactory = (IRecordDataParserFactory<?>) dataParserFactory;
- IRecordDataParser<?> dataParser = recordParserFactory.createRecordParser(ctx);
- // TODO(ali): revisit to think about passing data source name via setter or via createRecordParser
- dataParser.configure(recordReader.getDataSourceName(), recordReader.getLineNumber());
- dataParser.setValueEmbedder(valueEmbedder);
+ IRecordDataParser<?> dataParser = recordParserFactory.createRecordParser(runtimeContext);
if (isFeed) {
boolean isChangeFeed = ExternalDataUtils.isChangeFeed(configuration);
boolean isRecordWithMeta = ExternalDataUtils.isRecordWithMeta(configuration);
@@ -100,12 +95,10 @@
case STREAM:
IInputStreamFactory streamFactory = (IInputStreamFactory) dataSourceFactory;
AsterixInputStream stream = streamFactory.createInputStream(ctx, partition);
- stream.setValueEmbedder(valueEmbedder);
IStreamDataParserFactory streamParserFactory = (IStreamDataParserFactory) dataParserFactory;
// TODO(ali): revisit to think about passing data source name to parser
- IStreamDataParser streamParser = streamParserFactory.createInputStreamParser(ctx, partition);
+ IStreamDataParser streamParser = streamParserFactory.createInputStreamParser(runtimeContext);
streamParser.setInputStream(stream);
- streamParser.setValueEmbedder(valueEmbedder);
if (isFeed) {
return new FeedStreamDataFlowController(ctx, feedLogManager, streamParser, stream);
} else {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/DefaultExternalRuntimeDataContext.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/DefaultExternalRuntimeDataContext.java
new file mode 100644
index 0000000..9fe2ff7
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/DefaultExternalRuntimeDataContext.java
@@ -0,0 +1,67 @@
+/*
+ * 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.asterix.external.provider.context;
+
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
+import org.apache.asterix.external.input.filter.NoOpFilterValueEmbedder;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.asterix.external.util.ExternalDataConstants;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+
+public class DefaultExternalRuntimeDataContext implements IExternalDataRuntimeContext {
+ private final IHyracksTaskContext context;
+ private final int partition;
+
+ public DefaultExternalRuntimeDataContext(IHyracksTaskContext context) {
+ this(context, -1);
+ }
+
+ public DefaultExternalRuntimeDataContext(IHyracksTaskContext context, int partition) {
+ this.context = context;
+ this.partition = partition;
+ }
+
+ @Override
+ public final IHyracksTaskContext getTaskContext() {
+ return context;
+ }
+
+ @Override
+ public final int getPartition() {
+ return partition;
+ }
+
+ @Override
+ public IExternalFilterValueEmbedder getValueEmbedder() {
+ return NoOpFilterValueEmbedder.INSTANCE;
+ }
+
+ @Override
+ public Supplier<String> getDatasourceNameSupplier() {
+ return ExternalDataConstants.EMPTY_STRING;
+ }
+
+ @Override
+ public LongSupplier getLineNumberSupplier() {
+ return ExternalDataConstants.NO_LINES;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalReaderRuntimeDataContext.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalReaderRuntimeDataContext.java
new file mode 100644
index 0000000..587fc3e
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalReaderRuntimeDataContext.java
@@ -0,0 +1,65 @@
+/*
+ * 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.asterix.external.provider.context;
+
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.asterix.external.input.record.reader.stream.StreamRecordReader;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+
+public class ExternalReaderRuntimeDataContext extends ExternalStreamRuntimeDataContext {
+ private final IExternalFilterValueEmbedder valueEmbedder;
+ private StreamRecordReader reader;
+
+ public ExternalReaderRuntimeDataContext(IHyracksTaskContext context, int partition,
+ IExternalFilterValueEmbedder valueEmbedder) {
+ super(context, partition, valueEmbedder);
+ this.valueEmbedder = valueEmbedder;
+ reader = null;
+ }
+
+ @Override
+ public Supplier<String> getDatasourceNameSupplier() {
+ if (reader == null) {
+ // Safeguard: cannot invoke this method unless a reader has been set
+ throw new NullPointerException("Reader has not been set");
+ }
+ return reader.getDataSourceName();
+ }
+
+ @Override
+ public LongSupplier getLineNumberSupplier() {
+ if (reader == null) {
+ // Safeguard: cannot invoke this method unless a reader has been set
+ throw new NullPointerException("Reader has not been set");
+ }
+ return reader.getLineNumber();
+ }
+
+ @Override
+ public IExternalFilterValueEmbedder getValueEmbedder() {
+ return valueEmbedder;
+ }
+
+ public void setReader(StreamRecordReader reader) {
+ this.reader = reader;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalStreamRuntimeDataContext.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalStreamRuntimeDataContext.java
new file mode 100644
index 0000000..f60f2c8
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/ExternalStreamRuntimeDataContext.java
@@ -0,0 +1,37 @@
+/*
+ * 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.asterix.external.provider.context;
+
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+
+public class ExternalStreamRuntimeDataContext extends DefaultExternalRuntimeDataContext {
+ private final IExternalFilterValueEmbedder valueEmbedder;
+
+ public ExternalStreamRuntimeDataContext(IHyracksTaskContext context, int partition,
+ IExternalFilterValueEmbedder valueEmbedder) {
+ super(context, partition);
+ this.valueEmbedder = valueEmbedder;
+ }
+
+ @Override
+ public IExternalFilterValueEmbedder getValueEmbedder() {
+ return valueEmbedder;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/NoOpExternalRuntimeDataContext.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/NoOpExternalRuntimeDataContext.java
new file mode 100644
index 0000000..726ec23
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/context/NoOpExternalRuntimeDataContext.java
@@ -0,0 +1,60 @@
+/*
+ * 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.asterix.external.provider.context;
+
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
+import org.apache.asterix.external.input.filter.NoOpFilterValueEmbedder;
+import org.apache.asterix.external.input.filter.embedder.IExternalFilterValueEmbedder;
+import org.apache.asterix.external.util.ExternalDataConstants;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+
+public final class NoOpExternalRuntimeDataContext implements IExternalDataRuntimeContext {
+ public static final IExternalDataRuntimeContext INSTANCE = new NoOpExternalRuntimeDataContext();
+
+ private NoOpExternalRuntimeDataContext() {
+ }
+
+ @Override
+ public IHyracksTaskContext getTaskContext() {
+ throw new IllegalAccessError("should not be invoked");
+ }
+
+ @Override
+ public int getPartition() {
+ return -1;
+ }
+
+ @Override
+ public IExternalFilterValueEmbedder getValueEmbedder() {
+ return NoOpFilterValueEmbedder.INSTANCE;
+ }
+
+ @Override
+ public Supplier<String> getDatasourceNameSupplier() {
+ return ExternalDataConstants.EMPTY_STRING;
+ }
+
+ @Override
+ public LongSupplier getLineNumberSupplier() {
+ return ExternalDataConstants.NO_LINES;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
index 4f36915..bad5db7 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
@@ -25,6 +25,7 @@
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
@@ -32,7 +33,6 @@
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.IWarningCollector;
public class RecordWithPKTestReaderFactory implements IRecordReaderFactory<RecordWithPK<char[]>> {
@@ -56,8 +56,7 @@
}
@Override
- public IRecordReader<? extends RecordWithPK<char[]>> createRecordReader(final IHyracksTaskContext ctx,
- final int partition) {
+ public IRecordReader<? extends RecordWithPK<char[]>> createRecordReader(IExternalDataRuntimeContext context) {
return new TestAsterixMembersReader();
}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
index a21aeb6..7965d20 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
@@ -25,11 +25,11 @@
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.api.application.IServiceContext;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.IWarningCollector;
import com.couchbase.client.core.message.dcp.DCPRequest;
@@ -81,7 +81,8 @@
}
@Override
- public IRecordReader<? extends DCPRequest> createRecordReader(final IHyracksTaskContext ctx, final int partition) {
+ public IRecordReader<? extends DCPRequest> createRecordReader(IExternalDataRuntimeContext context) {
+ int partition = context.getPartition();
return new KVTestReader(partition, bucket, schedule,
(int) Math.ceil((double) numOfRecords / (double) numOfReaders), deleteCycle, upsertCycle,
(numOfRecords / numOfReaders) * partition);
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/factory/TestRecordWithPKParserFactory.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/factory/TestRecordWithPKParserFactory.java
index dbcfe0a..68059c1 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/factory/TestRecordWithPKParserFactory.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/factory/TestRecordWithPKParserFactory.java
@@ -23,6 +23,7 @@
import java.util.Map;
import java.util.TreeMap;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IRecordDataParserFactory;
import org.apache.asterix.external.input.record.RecordWithPK;
@@ -32,7 +33,6 @@
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.IAType;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@SuppressWarnings({ "unchecked" })
@@ -66,8 +66,9 @@
@SuppressWarnings("rawtypes")
@Override
- public IRecordDataParser<RecordWithPK<T>> createRecordParser(IHyracksTaskContext ctx) throws HyracksDataException {
- return new TestRecordWithPKParser(recordParserFactory.createRecordParser(ctx));
+ public IRecordDataParser<RecordWithPK<T>> createRecordParser(IExternalDataRuntimeContext context)
+ throws HyracksDataException {
+ return new TestRecordWithPKParser(recordParserFactory.createRecordParser(context));
}
@Override
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/test/RecordWithMetaTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/test/RecordWithMetaTest.java
index 888de08..07c4f88 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/test/RecordWithMetaTest.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/parser/test/RecordWithMetaTest.java
@@ -18,6 +18,9 @@
*/
package org.apache.asterix.external.parser.test;
+import org.apache.commons.io.FileUtils;
+import org.junit.Assert;
+
import java.io.File;
import java.io.PrintStream;
import java.nio.file.Files;
@@ -34,6 +37,7 @@
import org.apache.asterix.external.input.stream.LocalFSInputStream;
import org.apache.asterix.external.parser.ADMDataParser;
import org.apache.asterix.external.parser.RecordWithMetadataParser;
+import org.apache.asterix.external.provider.context.DefaultExternalRuntimeDataContext;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.FileSystemWatcher;
import org.apache.asterix.formats.nontagged.ADMPrinterFactoryProvider;
@@ -41,7 +45,6 @@
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.BuiltinType;
import org.apache.asterix.om.types.IAType;
-import org.apache.commons.io.FileUtils;
import org.apache.hyracks.algebricks.data.IPrinter;
import org.apache.hyracks.algebricks.data.IPrinterFactory;
import org.apache.hyracks.api.context.IHyracksTaskContext;
@@ -49,7 +52,6 @@
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import org.apache.hyracks.test.support.TestUtils;
-import org.junit.Assert;
public class RecordWithMetaTest {
private final IHyracksTaskContext ctx = TestUtils.createHyracksTask();
@@ -97,7 +99,7 @@
// create csv with json record reader
CSVToRecordWithMetadataAndPKConverter recordConverter =
new CSVToRecordWithMetadataAndPKConverter(valueIndex, delimiter, metaType, recordType, pkIndicators,
- pkIndexes, keyTypes, ctx.getWarningCollector());
+ pkIndexes, keyTypes, new DefaultExternalRuntimeDataContext(ctx));
// create the value parser <ADM in this case>
ADMDataParser valueParser = new ADMDataParser(recordType, false);
// create parser.
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSourceFactory.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSourceFactory.java
index 22bfb31..bcd0ab1 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSourceFactory.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSourceFactory.java
@@ -23,13 +23,13 @@
import java.util.Map;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
+import org.apache.asterix.external.api.IExternalDataRuntimeContext;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
import org.apache.asterix.metadata.api.IDatasourceFunction;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.IWarningCollector;
@@ -59,9 +59,9 @@
}
@Override
- public IRecordReader<? extends char[]> createRecordReader(IHyracksTaskContext ctx, int partition)
+ public IRecordReader<? extends char[]> createRecordReader(IExternalDataRuntimeContext context)
throws HyracksDataException {
- return function.createRecordReader(ctx, partition);
+ return function.createRecordReader(context.getTaskContext(), context.getPartition());
}
@Override
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17767
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Idde191ab6d99c48022bb65bec648e83e249e4b75
Gerrit-Change-Number: 17767
Gerrit-PatchSet: 1
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-MessageType: newchange