From 3b7bb2610f86e174d755844503e9acad85d54adf Mon Sep 17 00:00:00 2001 From: Cameron Lee Date: Mon, 29 Jul 2019 10:02:39 -0700 Subject: [PATCH] Revert "Fixing SamzaSQL Latency Metrics resolution (#1116)" This reverts commit ec536e14a1aa3d0fbd31c76418b605ea82c8b8d9. --- .../samza/system/IncomingMessageEnvelope.java | 8 ++--- .../samza/sql/SamzaSqlInputTransformer.java | 4 ++- .../samza/sql/avro/AvroRelConverter.java | 4 +-- .../samza/sql/data/SamzaSqlRelMessage.java | 6 ++-- .../sql/data/SamzaSqlRelMsgMetadata.java | 30 ++++++++--------- .../sql/translator/FilterTranslator.java | 16 +++++---- .../LogicalAggregateTranslator.java | 3 +- .../sql/translator/ProjectTranslator.java | 15 +++++---- .../samza/sql/translator/QueryTranslator.java | 31 ++++++++--------- .../samza/sql/translator/ScanTranslator.java | 17 +++++----- .../sql/translator/TranslatorConstants.java | 8 ++--- .../samza/sql/util/TranslatorUtils.java | 33 ------------------- .../sql/data/TestSamzaSqlRelMessage.java | 17 +++++----- .../TestSamzaSqlRelMessageSerde.java | 3 +- .../TestSamzaSqlRelRecordSerde.java | 2 +- .../sql/translator/TestFilterTranslator.java | 2 +- .../sql/translator/TestProjectTranslator.java | 2 +- .../TestSamzaSqlLocalTableJoinFunction.java | 16 ++++----- .../TestSamzaSqlRemoteTableJoinFunction.java | 2 +- 19 files changed, 94 insertions(+), 125 deletions(-) delete mode 100644 samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java diff --git a/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java b/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java index e8bb3315fb..73922536fa 100644 --- a/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java +++ b/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java @@ -37,10 +37,10 @@ public class IncomingMessageEnvelope { private final Object key; private final Object message; private final int size; - // the System's nanoTime when this event occurred, should be set by the event source, 0 means unassigned + // the timestamp when this event occured, should be set by the event source, 0 means unassigned private long eventTime = 0L; - // the System's nanoTime when this event is picked-up by samza, 0 means unassgined - private long arrivalTime; + // the timestamp when this event is pickedup by samza, 0 means unassgined + private long arrivalTime = 0L; /** * Constructs a new IncomingMessageEnvelope from specified components. @@ -70,7 +70,7 @@ public IncomingMessageEnvelope(SystemStreamPartition systemStreamPartition, Stri this.key = key; this.message = message; this.size = size; - this.arrivalTime = System.nanoTime(); + this.arrivalTime = Instant.now().toEpochMilli(); } /** diff --git a/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java b/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java index 6bcae8bf9e..2bbce22b9b 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java @@ -40,7 +40,9 @@ public class SamzaSqlInputTransformer implements InputTransformer { public Object apply(IncomingMessageEnvelope ime) { Assert.notNull(ime, "ime is null"); KV keyAndMessageKV = KV.of(ime.getKey(), ime.getMessage()); - SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata(ime.getEventTime(), ime.getArrivalTime(),0L); + SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata( + (ime.getEventTime() == 0) ? "" : Instant.ofEpochMilli(ime.getEventTime()).toString(), + (ime.getArrivalTime() == 0) ? "" : Instant.ofEpochMilli(ime.getArrivalTime()).toString(), null); SamzaSqlInputMessage samzaMsg = SamzaSqlInputMessage.of(keyAndMessageKV, metadata); return samzaMsg; } diff --git a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java index 3d3ebe24c8..fcb0c72d2d 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java @@ -96,7 +96,7 @@ public SamzaSqlRelMessage convertToRelMessage(KV samzaMessage) { } return new SamzaSqlRelMessage(samzaMessage.getKey(), payloadFieldNames, payloadFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); } /** @@ -108,7 +108,7 @@ public static SamzaSqlRelMessage convertToRelMessage(Object key, IndexedRecord r List payloadFieldValues = new ArrayList<>(); fetchFieldNamesAndValuesFromIndexedRecord(record, payloadFieldNames, payloadFieldValues, schema); return new SamzaSqlRelMessage(key, payloadFieldNames, payloadFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); } public static void fetchFieldNamesAndValuesFromIndexedRecord(IndexedRecord record, List fieldNames, diff --git a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java index 84db61f04c..d538a8e414 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java @@ -130,15 +130,15 @@ public Object getKey() { return key; } - public void setEventTime(long eventTime) { + public void setEventTime(String eventTime) { this.samzaSqlRelMsgMetadata.setEventTime(eventTime); } - public void setArrivalTime(long arrivalTime) { + public void setArrivalTime(String arrivalTime) { this.samzaSqlRelMsgMetadata.setArrivalTime(arrivalTime); } - public void setScanTime(long scanTime) { + public void setScanTime(String scanTime) { this.samzaSqlRelMsgMetadata.setScanTime(scanTime); } diff --git a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java index 60a4b11a64..14f28926f4 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java @@ -56,60 +56,60 @@ public class SamzaSqlRelMsgMetadata implements Serializable { * TODO: copy eventTime through from source to RelMessage */ @JsonProperty("eventTime") - private long eventTime; + private String eventTime; /** * the timestamp of when Samza App received the event * TODO: set arrivalTime during conversion from IME to SamzaMessage */ @JsonProperty("arrivalTime") - private long arrivalTime; + private String arrivalTime; /** * the timestamp when SamzaSQL query starts processing the event * set by the SamzaSQL Scan operator */ @JsonProperty("scanTime") - private long scanTime; + private String scanTime; - public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") long eventTime, @JsonProperty("arrivalTime") long arrivalTime, - @JsonProperty("scanTime") long scanTime) { + public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") String eventTime, @JsonProperty("arrivalTime") String arrivalTime, + @JsonProperty("scanTime") String scanTime) { this.eventTime = eventTime; this.arrivalTime = arrivalTime; this.scanTime = scanTime; } - public SamzaSqlRelMsgMetadata(long eventTime, long arrivalTime, long scanTime, boolean isNewInputMessage) { + public SamzaSqlRelMsgMetadata(String eventTime, String arrivalTime, String scanTime, boolean isNewInputMessage) { this(eventTime, arrivalTime, scanTime); this.isNewInputMessage = isNewInputMessage; } @JsonProperty("eventTime") - public long getEventTime() { return eventTime;} + public String getEventTime() { return eventTime;} - public void setEventTime(long eventTime) { + public void setEventTime(String eventTime) { this.eventTime = eventTime; } - public boolean hasEventTime() { return eventTime != 0; } + public boolean hasEventTime() { return eventTime != null && !eventTime.isEmpty(); } @JsonProperty("arrivalTime") - public long getarrivalTime() { return arrivalTime;} + public String getarrivalTime() { return arrivalTime;} - public void setArrivalTime(long arrivalTime) { + public void setArrivalTime(String arrivalTime) { this.arrivalTime = arrivalTime; } - public boolean hasArrivalTime() { return arrivalTime != 0; } + public boolean hasArrivalTime() { return arrivalTime != null && !arrivalTime.isEmpty(); } @JsonProperty("scanTime") - public long getscanTime() { return scanTime;} + public String getscanTime() { return scanTime;} - public void setScanTime(long scanTime) { + public void setScanTime(String scanTime) { this.scanTime = scanTime; } - public boolean hasScanTime() { return scanTime != 0; } + public boolean hasScanTime() { return scanTime != null && !scanTime.isEmpty(); } @JsonIgnore public void setIsSystemMessage(boolean isSystemMessage) { diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java index 18b73c82e4..ccab15a3f1 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java @@ -19,6 +19,9 @@ package org.apache.samza.sql.translator; +import com.google.common.annotations.VisibleForTesting; +import java.time.Duration; +import java.time.Instant; import java.util.Arrays; import java.util.Collections; import org.apache.calcite.rel.logical.LogicalFilter; @@ -32,7 +35,6 @@ import org.apache.samza.sql.data.Expression; import org.apache.samza.sql.data.SamzaSqlRelMessage; import org.apache.samza.sql.runner.SamzaSqlApplicationContext; -import org.apache.samza.sql.util.TranslatorUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -94,7 +96,7 @@ public void init(Context context) { @Override public boolean apply(SamzaSqlRelMessage message) { - long startProcessing = System.nanoTime(); + Instant startProcessing = Instant.now(); Object[] result = new Object[1]; expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(), message.getSamzaSqlRelRecord().getFieldValues().toArray(), result); @@ -103,7 +105,7 @@ public boolean apply(SamzaSqlRelMessage message) { log.debug( String.format("return value for input %s is %s", Arrays.asList(message.getSamzaSqlRelRecord().getFieldValues()).toString(), retVal)); - updateMetrics(startProcessing, retVal, System.nanoTime()); + updateMetrics(startProcessing, retVal, Instant.now()); return retVal; } else { log.error("return value is not boolean"); @@ -113,17 +115,17 @@ public boolean apply(SamzaSqlRelMessage message) { /** * Updates the MetricsRegistery of this operator - * @param startProcessing = nanoTime when processing of the message started - * @param endProcessing = nanoTIme when processing of the message ended + * @param startProcessing = begin processing of the message + * @param endProcessing = end of processing */ - private void updateMetrics(long startProcessing, boolean isOutput, long endProcessing) { + private void updateMetrics(Instant startProcessing, boolean isOutput, Instant endProcessing) { inputEvents.inc(); if (isOutput) { outputEvents.inc(); } else { filteredOutEvents.inc(); } - processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing)); + processingTime.update(Duration.between(startProcessing, endProcessing).toMillis()); } } diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java index cc8385796d..aaa6a9ef20 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java @@ -20,6 +20,7 @@ package org.apache.samza.sql.translator; import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.List; import org.apache.calcite.rel.logical.LogicalAggregate; @@ -80,7 +81,7 @@ void translate(final LogicalAggregate aggregate, final TranslatorContext context List fieldValues = windowPane.getKey().getKey().getSamzaSqlRelRecord().getFieldValues(); fieldNames.add(aggFieldNames.get(0)); fieldValues.add(windowPane.getMessage()); - return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata(0, 0,0)); + return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata("", "", "")); }); context.registerMessageStream(aggregate.getId(), outputStream); outputStream.map(new TranslatorOutputMetricsMapFunction(logicalOpId)); diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java index 0e49791c70..5b0dcc79f1 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java @@ -19,6 +19,8 @@ package org.apache.samza.sql.translator; +import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -42,7 +44,6 @@ import org.apache.samza.sql.data.SamzaSqlRelMessage; import org.apache.samza.sql.data.SamzaSqlRelMsgMetadata; import org.apache.samza.sql.runner.SamzaSqlApplicationContext; -import org.apache.samza.sql.util.TranslatorUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -112,7 +113,7 @@ public void init(Context context) { */ @Override public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) { - long arrivalTime = System.nanoTime(); + Instant arrivalTime = Instant.now(); RelDataType type = project.getRowType(); Object[] output = new Object[type.getFieldCount()]; expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(), @@ -121,22 +122,22 @@ public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) { for (int index = 0; index < output.length; index++) { names.add(index, project.getNamedProjects().get(index).getValue()); } - updateMetrics(arrivalTime, System.nanoTime(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage); + updateMetrics(arrivalTime, Instant.now(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage); return new SamzaSqlRelMessage(names, Arrays.asList(output), message.getSamzaSqlRelMsgMetadata()); } /** * Updates the Diagnostics Metrics (processing time and number of events) - * @param arrivalTime input message arrival nanoTime (= begin of processing in this operator) - * @param outputTime output message output nanoTime (=end of processing in this operator) + * @param arrivalTime input message arrival time (= beging of processing in this operator) + * @param outputTime output message output time (=end of processing in this operator) * @param isNewInputMessage whether the input Message is from new input message or not */ - private void updateMetrics(long arrivalTime, long outputTime, boolean isNewInputMessage) { + private void updateMetrics(Instant arrivalTime, Instant outputTime, boolean isNewInputMessage) { if (isNewInputMessage) { inputEvents.inc(); } outputEvents.inc(); - processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, outputTime)); + processingTime.update(Duration.between(arrivalTime, outputTime).toMillis()); } } diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java index 4e818e4ee2..2460b1f201 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java @@ -20,6 +20,8 @@ package org.apache.samza.sql.translator; import com.google.common.annotations.VisibleForTesting; +import java.time.Duration; +import java.time.Instant; import java.util.HashMap; import java.util.Map; import java.util.Optional; @@ -61,7 +63,6 @@ import org.apache.samza.sql.runner.SamzaSqlApplicationConfig; import org.apache.samza.sql.runner.SamzaSqlApplicationContext; import org.apache.samza.sql.util.SamzaSqlQueryParser; -import org.apache.samza.sql.util.TranslatorUtils; import org.apache.samza.system.descriptors.DelegatingSystemDescriptor; import org.apache.samza.system.descriptors.GenericOutputDescriptor; import org.apache.samza.table.Table; @@ -133,14 +134,14 @@ public void init(Context context) { queryLatency = new SamzaHistogram(metricsRegistry, queryLogicalId, TranslatorConstants.QUERY_LATENCY_NAME); queueingLatency = new SamzaHistogram(metricsRegistry, queryLogicalId, TranslatorConstants.QUEUEING_LATENCY_NAME); - + queryOutputEvents = metricsRegistry.newCounter(queryLogicalId, TranslatorConstants.OUTPUT_EVENTS_NAME); queryOutputEvents.clear(); } @Override public KV apply(SamzaSqlRelMessage message) { - long beginProcessing = System.nanoTime(); + Instant beginProcessing = Instant.now(); KV retKV = this.samzaMsgConverter.convertToSamzaMessage(message); if (message.getSamzaSqlRelRecord().containsField(SamzaSqlRelMessage.OP_NAME) && ((String) message.getSamzaSqlRelRecord().getField(SamzaSqlRelMessage.OP_NAME).get()).equalsIgnoreCase( @@ -148,36 +149,36 @@ public KV apply(SamzaSqlRelMessage message) { // If it is a delete op. Set the payload to null so that the record gets deleted. retKV = new KV<>(retKV.key, null); } - updateMetrics(beginProcessing, System.nanoTime(), message.getSamzaSqlRelMsgMetadata()); + updateMetrics(beginProcessing, Instant.now(), message.getSamzaSqlRelMsgMetadata()); return retKV; } /** * Updates the Diagnostics Metrics (processing time and number of events) - * @param beginProcessing when (nanoTime) sendOutput Started processing this message - * @param endProcessing when (nanoTime) sendOutput finished processing this message + * @param beginProcessing when sendOutput Started processing this message + * @param endProcessing when sendOutput finished processing this message * @param metadata the event's message metadata */ - private void updateMetrics(long beginProcessing, long endProcessing, SamzaSqlRelMsgMetadata metadata) { + private void updateMetrics(Instant beginProcessing, Instant endProcessing, SamzaSqlRelMsgMetadata metadata) { /* insert (SendToOutputStream) metrics */ - insertProcessingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(beginProcessing, endProcessing)); + insertProcessingTime.update(Duration.between(beginProcessing, endProcessing).toMillis()); /* query metrics */ - long outputTime = System.nanoTime(); + Instant outputTime = Instant.now(); queryOutputEvents.inc(); /* TODO: remove scanTime validation once code to assign it is stable */ Validate.isTrue(metadata.hasScanTime()); - long scanTime = metadata.getscanTime(); - queryLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(scanTime, outputTime)); + Instant scanTime = Instant.parse(metadata.getscanTime()); + queryLatency.update(Duration.between(scanTime, outputTime).toMillis()); /** TODO: change if hasArrivalTime to validation once arrivalTime is assigned, and later remove the check once code is stable */ if (metadata.hasArrivalTime()) { - long arrivalTime = metadata.getarrivalTime(); - queueingLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, scanTime)); + Instant arrivalTime = Instant.parse(metadata.getarrivalTime()); + queueingLatency.update(Duration.between(arrivalTime, scanTime).toMillis()); } /* since availability of eventTime depends on source, we need the following check */ if (metadata.hasEventTime()) { - long eventTime = metadata.getEventTime(); - totalLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(eventTime, outputTime)); + Instant eventTime = Instant.parse(metadata.getEventTime()); + totalLatency.update(Duration.between(eventTime, outputTime).toMillis()); } } } // OutputMapFunction diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java index ceee6d261a..5fa04d82d3 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java @@ -19,6 +19,8 @@ package org.apache.samza.sql.translator; +import java.time.Duration; +import java.time.Instant; import java.util.List; import java.util.Map; import org.apache.calcite.rel.core.TableScan; @@ -40,7 +42,6 @@ import org.apache.samza.sql.interfaces.SamzaRelConverter; import org.apache.samza.sql.interfaces.SqlIOConfig; import org.apache.samza.sql.runner.SamzaSqlApplicationContext; -import org.apache.samza.sql.util.TranslatorUtils; import org.apache.samza.system.descriptors.DelegatingSystemDescriptor; import org.apache.samza.system.descriptors.InputDescriptor; import org.apache.samza.system.descriptors.InputTransformer; @@ -127,25 +128,25 @@ public void init(Context context) { @Override public SamzaSqlRelMessage apply(SamzaSqlInputMessage samzaSqlInputMessage) { - long startProcessing = System.nanoTime(); + Instant startProcessing = Instant.now(); /* SAMZA-2089/LISAMZA-10654: the SamzaRelConverter.convertToRelMessage currently does not initialize * the samzaSqlRelMessage.samzaSqlRelMsgMetadata, this needs to be fixed */ SamzaSqlRelMessage retMsg = this.msgConverter.convertToRelMessage(samzaSqlInputMessage.getKeyAndMessageKV()); retMsg.setEventTime(samzaSqlInputMessage.getMetadata().getEventTime()); retMsg.setArrivalTime(samzaSqlInputMessage.getMetadata().getarrivalTime()); - retMsg.setScanTime(startProcessing); - updateMetrics(startProcessing, System.nanoTime()); + retMsg.setScanTime(startProcessing.toString()); + updateMetrics(startProcessing, Instant.now()); return retMsg; } /** * Updates the MetricsRegistery of this operator - * @param startProcessing = begin (nanoTime) processing of the message - * @param endProcessing = end (nanoTime) of processing + * @param startProcessing = begin processing of the message + * @param endProcessing = end of processing */ - private void updateMetrics(long startProcessing, long endProcessing) { + private void updateMetrics(Instant startProcessing, Instant endProcessing) { queryInputEvents.inc(); - processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing)); + processingTime.update(Duration.between(startProcessing, endProcessing).toMillis()); } } // ScanMapFunction diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java index d788297c3b..c2f9ea5f69 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java @@ -20,10 +20,10 @@ package org.apache.samza.sql.translator; public class TranslatorConstants { - public static final String PROCESSING_TIME_NAME = "processingTimeMicros"; - public static final String TOTAL_LATENCY_NAME = "totalLatencyMicros"; - public static final String QUERY_LATENCY_NAME = "queryLatencyMicros"; - public static final String QUEUEING_LATENCY_NAME = "queueingLatencyMicros"; + public static final String PROCESSING_TIME_NAME = "processingTimeMs"; + public static final String TOTAL_LATENCY_NAME = "totalLatencyMs"; + public static final String QUERY_LATENCY_NAME = "queryLatencyMs"; + public static final String QUEUEING_LATENCY_NAME = "queueingLatencyMs"; public static final String INPUT_EVENTS_NAME = "inputEvents"; public static final String FILTERED_EVENTS_NAME = "filteredEvents"; public static final String OUTPUT_EVENTS_NAME = "outputEvents"; diff --git a/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java b/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java deleted file mode 100644 index c9d0239611..0000000000 --- a/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.samza.sql.util; - -public class TranslatorUtils { - static long MICROS_IN_NANO = 1000; - /** - * computes duraiton and converts it to micro resolution - * @param start start nanoTime - * @param end end nanoTime - * @return duration in micro resolution - */ - static public long getMicroDurationFromNanoTimes(long start, long end) { - return (end - start) / MICROS_IN_NANO; - } -} diff --git a/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java b/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java index ac8e10dbe2..dffb77e3a1 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java @@ -36,34 +36,33 @@ public class TestSamzaSqlRelMessage { @Test public void testGetField() { - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); Assert.assertEquals(values.get(0), message.getSamzaSqlRelRecord().getField(names.get(0)).get()); Assert.assertEquals(values.get(1), message.getSamzaSqlRelRecord().getField(names.get(1)).get()); } @Test public void testGetNonExistentField() { - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); Assert.assertFalse(message.getSamzaSqlRelRecord().getField("field3").isPresent()); } @Test public void testEquality() { - SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlRelMessage message2 = new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value1", "value2"), - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); Assert.assertEquals(message1, message2); Assert.assertEquals(message1.hashCode(), message2.hashCode()); } @Test public void testInEquality() { - SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlRelMessage message2 = new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value2", "value2"), - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); Assert.assertNotEquals(message1, message2); Assert.assertNotEquals(message1.hashCode(), message2.hashCode()); } @@ -71,7 +70,7 @@ public void testInEquality() { @Test public void testCompositeKeyCreation() { List keyPartNames = Arrays.asList("kfield1", "kfield2"); - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlRelRecord relRecord1 = SamzaSqlRelMessage.createSamzaSqlCompositeKey(message, Collections.singletonList(0)); Assert.assertEquals(relRecord1.getFieldNames().size(), 1); @@ -90,7 +89,7 @@ public void testCompositeKeyCreation() { @Test (expected = IllegalArgumentException.class) public void testCompositeKeyCreationWithInEqualKeyNameValues() { List keyPartNames = Arrays.asList("kfield1", "kfield2"); - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlRelRecord relRecord1 = SamzaSqlRelMessage.createSamzaSqlCompositeKey(message, Arrays.asList(1, 0), SamzaSqlRelMessage.getSamzaSqlCompositeKeyFieldNames(keyPartNames, Arrays.asList(1))); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java index 689e8d2d29..5afa4fe80c 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java @@ -53,8 +53,7 @@ public class TestSamzaSqlRelMessageSerde { @Test public void testWithDifferentFields() { - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlRelMessageSerde serde = (SamzaSqlRelMessageSerde) new SamzaSqlRelMessageSerdeFactory().getSerde(null, null); SamzaSqlRelMessage resultMsg = serde.fromBytes(serde.toBytes(message)); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java index 6885fc1acf..4e6ff4f8b9 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java @@ -51,7 +51,7 @@ public class TestSamzaSqlRelRecordSerde { @Test public void testWithDifferentFields() { SamzaSqlRelRecord record = - new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0,0, 0)).getSamzaSqlRelRecord(); + new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")).getSamzaSqlRelRecord(); SamzaSqlRelRecordSerde serde = (SamzaSqlRelRecordSerde) new SamzaSqlRelRecordSerdeFactory().getSerde(null, null); SamzaSqlRelRecord resultRecord = serde.fromBytes(serde.toBytes(record)); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java index be64cc4b84..2a98c8c9e9 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java @@ -125,7 +125,7 @@ public void testTranslate() throws IOException, ClassNotFoundException { // Calling filterFn.apply() to verify the filter function is correctly applied to the input message SamzaSqlRelMessage mockInputMsg = new SamzaSqlRelMessage(new ArrayList<>(), new ArrayList<>(), - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlExecutionContext executionContext = mock(SamzaSqlExecutionContext.class); DataContext dataContext = mock(DataContext.class); when(mockTranslatorContext.getExecutionContext()).thenReturn(executionContext); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java index 829532cb52..3a2b943f28 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java @@ -143,7 +143,7 @@ public void testTranslate() throws IOException, ClassNotFoundException { // Calling mapFn.apply() to verify the filter function is correctly applied to the input message SamzaSqlRelMessage mockInputMsg = new SamzaSqlRelMessage(new ArrayList<>(), new ArrayList<>(), - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlExecutionContext executionContext = mock(SamzaSqlExecutionContext.class); DataContext dataContext = mock(DataContext.class); when(mockTranslatorContext.getExecutionContext()).thenReturn(executionContext); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java index b92dfe8868..53c7bc14cb 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java @@ -44,10 +44,8 @@ public class TestSamzaSqlLocalTableJoinFunction { @Test public void testWithInnerJoinWithTableOnRight() { - SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); - SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(0, 1); List tableKeyIds = Arrays.asList(0, 1); @@ -79,10 +77,8 @@ public void testWithInnerJoinWithTableOnRight() { @Test public void testWithInnerJoinWithTableOnLeft() { - SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); - SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(0, 2); List tableKeyIds = Arrays.asList(0, 2); @@ -115,7 +111,7 @@ public void testWithInnerJoinWithTableOnLeft() { @Test public void testNullRecordWithInnerJoin() { SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(0, 1); List tableKeyIds = Arrays.asList(2, 3); @@ -139,7 +135,7 @@ public void testNullRecordWithInnerJoin() { @Test public void testNullRecordWithLeftOuterJoin() { SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); JoinRelType joinRelType = JoinRelType.LEFT; List streamKeyIds = Arrays.asList(0, 1); List tableKeyIds = Arrays.asList(2, 3); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java index 20cea14c30..2deb6ad942 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java @@ -69,7 +69,7 @@ public void testWithInnerJoinWithTableOnRight() { tableRecord.put("name", "name1"); SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata(0, 0, 0)); + new SamzaSqlRelMsgMetadata("", "", "")); SamzaSqlRelMessage tableMsg = relConverter.convertToRelMessage(new KV(tableRecord.get("id"), tableRecord)); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(1);