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 2021/03/22 23:49:09 UTC
Change in asterixdb[mad-hatter]: WIP: external dataset perf
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/10643 )
Change subject: WIP: external dataset perf
......................................................................
WIP: external dataset perf
Change-Id: I7aa889527d95f28eec29edbfee34e6c95d23f64a
---
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
6 files changed, 38 insertions(+), 8 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/43/10643/1
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
index 2c19f9d..646990b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
@@ -32,6 +32,7 @@
protected final IRecordDataParser<T> dataParser;
protected final IRecordReader<? extends T> recordReader;
protected final int numOfTupleFields;
+ protected boolean skipParse;
public RecordDataFlowController(IHyracksTaskContext ctx, IRecordDataParser<T> dataParser,
IRecordReader<? extends T> recordReader, int numOfTupleFields) {
@@ -39,6 +40,7 @@
this.dataParser = dataParser;
this.recordReader = recordReader;
this.numOfTupleFields = numOfTupleFields;
+ this.skipParse = Boolean.getBoolean("ext.skip.parse");
}
@Override
@@ -50,11 +52,13 @@
while (recordReader.hasNext()) {
IRawRecord<? extends T> record = recordReader.next();
tb.reset();
- if (dataParser.parse(record, tb.getDataOutput())) {
- tb.addFieldEndOffset();
- appendOtherTupleFields(tb);
- tupleForwarder.addTuple(tb);
- processedTuples++;
+ if (!skipParse) {
+ if (dataParser.parse(record, tb.getDataOutput())) {
+ tb.addFieldEndOffset();
+ appendOtherTupleFields(tb);
+ tupleForwarder.addTuple(tb);
+ processedTuples++;
+ }
}
}
tupleForwarder.complete();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
index f558820..2f056af 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
@@ -29,10 +29,12 @@
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
import org.apache.asterix.external.input.stream.AbstractMultipleInputStream;
+import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.util.CleanupUtils;
import org.apache.hyracks.util.LogRedactionUtil;
+import org.apache.hyracks.util.StorageUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -59,7 +61,9 @@
this.configuration = configuration;
this.filePaths = filePaths;
this.s3Client = buildAwsS3Client(configuration);
- this.bufferSize = ExternalDataUtils.getOrDefaultBufferSize(configuration);
+ String buffSize = System.getProperty("ext.buffer.size");
+ this.bufferSize =
+ buffSize != null ? (int) StorageUtil.getByteValue(buffSize) : ExternalDataConstants.DEFAULT_BUFFER_SIZE;
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
index 2ff5cfa..b3f4c2b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
@@ -66,6 +66,8 @@
private State state = State.TOP_LEVEL;
private long beginLineNumber = 1;
private long lineNumber = 1;
+ private boolean skipParse;
+ private static final char[] EMPTY = "{}".toCharArray();
private static final List<String> recordReaderFormats = Collections.unmodifiableList(
Arrays.asList(ExternalDataConstants.FORMAT_ADM, ExternalDataConstants.FORMAT_JSON_LOWER_CASE,
@@ -85,6 +87,7 @@
if (recordStart == recordEnd) {
throw new RuntimeDataException(ErrorCode.INVALID_REQ_PARAM_VAL, KEY_RECORD_END, recordEnd);
}
+ skipParse = Boolean.getBoolean("ext.skip.stream.parse");
}
@Override
@@ -114,6 +117,16 @@
}
resetForNewRecord();
beginLineNumber = lineNumber;
+ if (skipParse) {
+ bufferLength = reader.read(inputBuffer);
+ if (bufferLength < 0) {
+ close();
+ return false; // EOF
+ }
+ record.append(EMPTY);
+ record.endRecord();
+ return true;
+ }
do {
int startPosn = bufferPosn; // starting from where we left off the last time
if (bufferPosn >= bufferLength) {
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 1e003a1..0369911 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
@@ -33,10 +33,12 @@
import org.apache.asterix.external.dataflow.AbstractFeedDataFlowController;
import org.apache.asterix.external.input.record.CharArrayRecord;
import org.apache.asterix.external.input.stream.AsterixInputStreamReader;
+import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.external.util.FeedLogManager;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.util.StorageUtil;
public abstract class StreamRecordReader implements IRecordReader<char[]>, IStreamNotificationHandler {
protected AsterixInputStreamReader reader;
@@ -50,7 +52,9 @@
private Supplier<String> previousDataSourceName = EMPTY_STRING;
public void configure(AsterixInputStream inputStream, Map<String, String> config) {
- int bufferSize = ExternalDataUtils.getOrDefaultBufferSize(config);
+ String buffSize = System.getProperty("ext.buffer.size");
+ int bufferSize =
+ buffSize != null ? (int) StorageUtil.getByteValue(buffSize) : ExternalDataConstants.DEFAULT_BUFFER_SIZE;
this.reader = new AsterixInputStreamReader(inputStream, bufferSize);
record = new CharArrayRecord();
inputBuffer = new char[bufferSize];
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
index ddbd350..67fb73a 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
@@ -45,6 +45,9 @@
this.byteBuffer = ByteBuffer.wrap(bytes);
this.charBuffer = CharBuffer.allocate(bufferSize);
this.decoder = StandardCharsets.UTF_8.newDecoder();
+ bytes = new byte[bufferSize];
+ byteBuffer = ByteBuffer.wrap(bytes);
+ charBuffer = CharBuffer.allocate(bufferSize);
this.byteBuffer.flip();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 1911083..22245a9 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -24,6 +24,8 @@
import java.util.function.LongSupplier;
import java.util.function.Supplier;
+import org.apache.hyracks.util.StorageUtil;
+
public class ExternalDataConstants {
private ExternalDataConstants() {
@@ -251,7 +253,7 @@
/**
* Size default values
*/
- public static final int DEFAULT_BUFFER_SIZE = 4096;
+ public static final int DEFAULT_BUFFER_SIZE = StorageUtil.getIntSizeInBytes(8, StorageUtil.StorageUnit.KILOBYTE);
public static final float DEFAULT_BUFFER_INCREMENT_FACTOR = 1.5F;
public static final int DEFAULT_QUEUE_SIZE = 64;
public static final int MAX_RECORD_SIZE = 32000000;
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/10643
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: mad-hatter
Gerrit-Change-Id: I7aa889527d95f28eec29edbfee34e6c95d23f64a
Gerrit-Change-Number: 10643
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>
Gerrit-MessageType: newchange
Change in asterixdb[mad-hatter]: WIP: external dataset perf
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
Anon. E. Moose #1000171 has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/10643 )
Change subject: WIP: external dataset perf
......................................................................
Patch Set 1: Contrib-2
Analytics Compatibility Compilation Failed
https://cbjenkins.page.link/fpE6NBMgiaw5Getk9 : UNSTABLE
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/10643
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: mad-hatter
Gerrit-Change-Id: I7aa889527d95f28eec29edbfee34e6c95d23f64a
Gerrit-Change-Number: 10643
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-CC: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Comment-Date: Mon, 22 Mar 2021 23:53:08 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment
Change in asterixdb[mad-hatter]: WIP: external dataset perf
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/10643 )
Change subject: WIP: external dataset perf
......................................................................
WIP: external dataset perf
Change-Id: I7aa889527d95f28eec29edbfee34e6c95d23f64a
---
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
6 files changed, 38 insertions(+), 8 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/43/10643/1
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
index 2c19f9d..646990b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/RecordDataFlowController.java
@@ -32,6 +32,7 @@
protected final IRecordDataParser<T> dataParser;
protected final IRecordReader<? extends T> recordReader;
protected final int numOfTupleFields;
+ protected boolean skipParse;
public RecordDataFlowController(IHyracksTaskContext ctx, IRecordDataParser<T> dataParser,
IRecordReader<? extends T> recordReader, int numOfTupleFields) {
@@ -39,6 +40,7 @@
this.dataParser = dataParser;
this.recordReader = recordReader;
this.numOfTupleFields = numOfTupleFields;
+ this.skipParse = Boolean.getBoolean("ext.skip.parse");
}
@Override
@@ -50,11 +52,13 @@
while (recordReader.hasNext()) {
IRawRecord<? extends T> record = recordReader.next();
tb.reset();
- if (dataParser.parse(record, tb.getDataOutput())) {
- tb.addFieldEndOffset();
- appendOtherTupleFields(tb);
- tupleForwarder.addTuple(tb);
- processedTuples++;
+ if (!skipParse) {
+ if (dataParser.parse(record, tb.getDataOutput())) {
+ tb.addFieldEndOffset();
+ appendOtherTupleFields(tb);
+ tupleForwarder.addTuple(tb);
+ processedTuples++;
+ }
}
}
tupleForwarder.complete();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
index f558820..2f056af 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStream.java
@@ -29,10 +29,12 @@
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
import org.apache.asterix.external.input.stream.AbstractMultipleInputStream;
+import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.util.CleanupUtils;
import org.apache.hyracks.util.LogRedactionUtil;
+import org.apache.hyracks.util.StorageUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -59,7 +61,9 @@
this.configuration = configuration;
this.filePaths = filePaths;
this.s3Client = buildAwsS3Client(configuration);
- this.bufferSize = ExternalDataUtils.getOrDefaultBufferSize(configuration);
+ String buffSize = System.getProperty("ext.buffer.size");
+ this.bufferSize =
+ buffSize != null ? (int) StorageUtil.getByteValue(buffSize) : ExternalDataConstants.DEFAULT_BUFFER_SIZE;
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
index 2ff5cfa..b3f4c2b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/SemiStructuredRecordReader.java
@@ -66,6 +66,8 @@
private State state = State.TOP_LEVEL;
private long beginLineNumber = 1;
private long lineNumber = 1;
+ private boolean skipParse;
+ private static final char[] EMPTY = "{}".toCharArray();
private static final List<String> recordReaderFormats = Collections.unmodifiableList(
Arrays.asList(ExternalDataConstants.FORMAT_ADM, ExternalDataConstants.FORMAT_JSON_LOWER_CASE,
@@ -85,6 +87,7 @@
if (recordStart == recordEnd) {
throw new RuntimeDataException(ErrorCode.INVALID_REQ_PARAM_VAL, KEY_RECORD_END, recordEnd);
}
+ skipParse = Boolean.getBoolean("ext.skip.stream.parse");
}
@Override
@@ -114,6 +117,16 @@
}
resetForNewRecord();
beginLineNumber = lineNumber;
+ if (skipParse) {
+ bufferLength = reader.read(inputBuffer);
+ if (bufferLength < 0) {
+ close();
+ return false; // EOF
+ }
+ record.append(EMPTY);
+ record.endRecord();
+ return true;
+ }
do {
int startPosn = bufferPosn; // starting from where we left off the last time
if (bufferPosn >= bufferLength) {
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 1e003a1..0369911 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
@@ -33,10 +33,12 @@
import org.apache.asterix.external.dataflow.AbstractFeedDataFlowController;
import org.apache.asterix.external.input.record.CharArrayRecord;
import org.apache.asterix.external.input.stream.AsterixInputStreamReader;
+import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.external.util.FeedLogManager;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.util.StorageUtil;
public abstract class StreamRecordReader implements IRecordReader<char[]>, IStreamNotificationHandler {
protected AsterixInputStreamReader reader;
@@ -50,7 +52,9 @@
private Supplier<String> previousDataSourceName = EMPTY_STRING;
public void configure(AsterixInputStream inputStream, Map<String, String> config) {
- int bufferSize = ExternalDataUtils.getOrDefaultBufferSize(config);
+ String buffSize = System.getProperty("ext.buffer.size");
+ int bufferSize =
+ buffSize != null ? (int) StorageUtil.getByteValue(buffSize) : ExternalDataConstants.DEFAULT_BUFFER_SIZE;
this.reader = new AsterixInputStreamReader(inputStream, bufferSize);
record = new CharArrayRecord();
inputBuffer = new char[bufferSize];
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
index ddbd350..67fb73a 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
@@ -45,6 +45,9 @@
this.byteBuffer = ByteBuffer.wrap(bytes);
this.charBuffer = CharBuffer.allocate(bufferSize);
this.decoder = StandardCharsets.UTF_8.newDecoder();
+ bytes = new byte[bufferSize];
+ byteBuffer = ByteBuffer.wrap(bytes);
+ charBuffer = CharBuffer.allocate(bufferSize);
this.byteBuffer.flip();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 1911083..22245a9 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -24,6 +24,8 @@
import java.util.function.LongSupplier;
import java.util.function.Supplier;
+import org.apache.hyracks.util.StorageUtil;
+
public class ExternalDataConstants {
private ExternalDataConstants() {
@@ -251,7 +253,7 @@
/**
* Size default values
*/
- public static final int DEFAULT_BUFFER_SIZE = 4096;
+ public static final int DEFAULT_BUFFER_SIZE = StorageUtil.getIntSizeInBytes(8, StorageUtil.StorageUnit.KILOBYTE);
public static final float DEFAULT_BUFFER_INCREMENT_FACTOR = 1.5F;
public static final int DEFAULT_QUEUE_SIZE = 64;
public static final int MAX_RECORD_SIZE = 32000000;
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/10643
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: mad-hatter
Gerrit-Change-Id: I7aa889527d95f28eec29edbfee34e6c95d23f64a
Gerrit-Change-Number: 10643
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>
Gerrit-MessageType: newchange