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 73922536fa..e8bb3315fb 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 timestamp when this event occured, should be set by the event source, 0 means unassigned + // the System's nanoTime when this event occurred, should be set by the event source, 0 means unassigned private long eventTime = 0L; - // the timestamp when this event is pickedup by samza, 0 means unassgined - private long arrivalTime = 0L; + // the System's nanoTime when this event is picked-up by samza, 0 means unassgined + private long arrivalTime; /** * 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 = Instant.now().toEpochMilli(); + this.arrivalTime = System.nanoTime(); } /** 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 2bbce22b9b..6bcae8bf9e 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,9 +40,7 @@ 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() == 0) ? "" : Instant.ofEpochMilli(ime.getEventTime()).toString(), - (ime.getArrivalTime() == 0) ? "" : Instant.ofEpochMilli(ime.getArrivalTime()).toString(), null); + SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata(ime.getEventTime(), ime.getArrivalTime(),0L); 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 fcb0c72d2d..3d3ebe24c8 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("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); } /** @@ -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("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); } 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 d538a8e414..84db61f04c 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(String eventTime) { + public void setEventTime(long eventTime) { this.samzaSqlRelMsgMetadata.setEventTime(eventTime); } - public void setArrivalTime(String arrivalTime) { + public void setArrivalTime(long arrivalTime) { this.samzaSqlRelMsgMetadata.setArrivalTime(arrivalTime); } - public void setScanTime(String scanTime) { + public void setScanTime(long 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 14f28926f4..60a4b11a64 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 String eventTime; + private long eventTime; /** * the timestamp of when Samza App received the event * TODO: set arrivalTime during conversion from IME to SamzaMessage */ @JsonProperty("arrivalTime") - private String arrivalTime; + private long arrivalTime; /** * the timestamp when SamzaSQL query starts processing the event * set by the SamzaSQL Scan operator */ @JsonProperty("scanTime") - private String scanTime; + private long scanTime; - public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") String eventTime, @JsonProperty("arrivalTime") String arrivalTime, - @JsonProperty("scanTime") String scanTime) { + public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") long eventTime, @JsonProperty("arrivalTime") long arrivalTime, + @JsonProperty("scanTime") long scanTime) { this.eventTime = eventTime; this.arrivalTime = arrivalTime; this.scanTime = scanTime; } - public SamzaSqlRelMsgMetadata(String eventTime, String arrivalTime, String scanTime, boolean isNewInputMessage) { + public SamzaSqlRelMsgMetadata(long eventTime, long arrivalTime, long scanTime, boolean isNewInputMessage) { this(eventTime, arrivalTime, scanTime); this.isNewInputMessage = isNewInputMessage; } @JsonProperty("eventTime") - public String getEventTime() { return eventTime;} + public long getEventTime() { return eventTime;} - public void setEventTime(String eventTime) { + public void setEventTime(long eventTime) { this.eventTime = eventTime; } - public boolean hasEventTime() { return eventTime != null && !eventTime.isEmpty(); } + public boolean hasEventTime() { return eventTime != 0; } @JsonProperty("arrivalTime") - public String getarrivalTime() { return arrivalTime;} + public long getarrivalTime() { return arrivalTime;} - public void setArrivalTime(String arrivalTime) { + public void setArrivalTime(long arrivalTime) { this.arrivalTime = arrivalTime; } - public boolean hasArrivalTime() { return arrivalTime != null && !arrivalTime.isEmpty(); } + public boolean hasArrivalTime() { return arrivalTime != 0; } @JsonProperty("scanTime") - public String getscanTime() { return scanTime;} + public long getscanTime() { return scanTime;} - public void setScanTime(String scanTime) { + public void setScanTime(long scanTime) { this.scanTime = scanTime; } - public boolean hasScanTime() { return scanTime != null && !scanTime.isEmpty(); } + public boolean hasScanTime() { return scanTime != 0; } @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 ccab15a3f1..18b73c82e4 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,9 +19,6 @@ 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; @@ -35,6 +32,7 @@ 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; @@ -96,7 +94,7 @@ public void init(Context context) { @Override public boolean apply(SamzaSqlRelMessage message) { - Instant startProcessing = Instant.now(); + long startProcessing = System.nanoTime(); Object[] result = new Object[1]; expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(), message.getSamzaSqlRelRecord().getFieldValues().toArray(), result); @@ -105,7 +103,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, Instant.now()); + updateMetrics(startProcessing, retVal, System.nanoTime()); return retVal; } else { log.error("return value is not boolean"); @@ -115,17 +113,17 @@ public boolean apply(SamzaSqlRelMessage message) { /** * Updates the MetricsRegistery of this operator - * @param startProcessing = begin processing of the message - * @param endProcessing = end of processing + * @param startProcessing = nanoTime when processing of the message started + * @param endProcessing = nanoTIme when processing of the message ended */ - private void updateMetrics(Instant startProcessing, boolean isOutput, Instant endProcessing) { + private void updateMetrics(long startProcessing, boolean isOutput, long endProcessing) { inputEvents.inc(); if (isOutput) { outputEvents.inc(); } else { filteredOutEvents.inc(); } - processingTime.update(Duration.between(startProcessing, endProcessing).toMillis()); + processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing)); } } 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 aaa6a9ef20..cc8385796d 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,7 +20,6 @@ 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; @@ -81,7 +80,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("", "", "")); + return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata(0, 0,0)); }); 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 5b0dcc79f1..0e49791c70 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,8 +19,6 @@ 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; @@ -44,6 +42,7 @@ 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; @@ -113,7 +112,7 @@ public void init(Context context) { */ @Override public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) { - Instant arrivalTime = Instant.now(); + long arrivalTime = System.nanoTime(); RelDataType type = project.getRowType(); Object[] output = new Object[type.getFieldCount()]; expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(), @@ -122,22 +121,22 @@ public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) { for (int index = 0; index < output.length; index++) { names.add(index, project.getNamedProjects().get(index).getValue()); } - updateMetrics(arrivalTime, Instant.now(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage); + updateMetrics(arrivalTime, System.nanoTime(), 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 time (= beging of processing in this operator) - * @param outputTime output message output time (=end of processing in this operator) + * @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 isNewInputMessage whether the input Message is from new input message or not */ - private void updateMetrics(Instant arrivalTime, Instant outputTime, boolean isNewInputMessage) { + private void updateMetrics(long arrivalTime, long outputTime, boolean isNewInputMessage) { if (isNewInputMessage) { inputEvents.inc(); } outputEvents.inc(); - processingTime.update(Duration.between(arrivalTime, outputTime).toMillis()); + processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, outputTime)); } } 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 2460b1f201..4e818e4ee2 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,8 +20,6 @@ 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; @@ -63,6 +61,7 @@ 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; @@ -134,14 +133,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) { - Instant beginProcessing = Instant.now(); + long beginProcessing = System.nanoTime(); KV retKV = this.samzaMsgConverter.convertToSamzaMessage(message); if (message.getSamzaSqlRelRecord().containsField(SamzaSqlRelMessage.OP_NAME) && ((String) message.getSamzaSqlRelRecord().getField(SamzaSqlRelMessage.OP_NAME).get()).equalsIgnoreCase( @@ -149,36 +148,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, Instant.now(), message.getSamzaSqlRelMsgMetadata()); + updateMetrics(beginProcessing, System.nanoTime(), message.getSamzaSqlRelMsgMetadata()); return retKV; } /** * Updates the Diagnostics Metrics (processing time and number of events) - * @param beginProcessing when sendOutput Started processing this message - * @param endProcessing when sendOutput finished processing this message + * @param beginProcessing when (nanoTime) sendOutput Started processing this message + * @param endProcessing when (nanoTime) sendOutput finished processing this message * @param metadata the event's message metadata */ - private void updateMetrics(Instant beginProcessing, Instant endProcessing, SamzaSqlRelMsgMetadata metadata) { + private void updateMetrics(long beginProcessing, long endProcessing, SamzaSqlRelMsgMetadata metadata) { /* insert (SendToOutputStream) metrics */ - insertProcessingTime.update(Duration.between(beginProcessing, endProcessing).toMillis()); + insertProcessingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(beginProcessing, endProcessing)); /* query metrics */ - Instant outputTime = Instant.now(); + long outputTime = System.nanoTime(); queryOutputEvents.inc(); /* TODO: remove scanTime validation once code to assign it is stable */ Validate.isTrue(metadata.hasScanTime()); - Instant scanTime = Instant.parse(metadata.getscanTime()); - queryLatency.update(Duration.between(scanTime, outputTime).toMillis()); + long scanTime = metadata.getscanTime(); + queryLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(scanTime, outputTime)); /** TODO: change if hasArrivalTime to validation once arrivalTime is assigned, and later remove the check once code is stable */ if (metadata.hasArrivalTime()) { - Instant arrivalTime = Instant.parse(metadata.getarrivalTime()); - queueingLatency.update(Duration.between(arrivalTime, scanTime).toMillis()); + long arrivalTime = metadata.getarrivalTime(); + queueingLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, scanTime)); } /* since availability of eventTime depends on source, we need the following check */ if (metadata.hasEventTime()) { - Instant eventTime = Instant.parse(metadata.getEventTime()); - totalLatency.update(Duration.between(eventTime, outputTime).toMillis()); + long eventTime = metadata.getEventTime(); + totalLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(eventTime, outputTime)); } } } // 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 5fa04d82d3..ceee6d261a 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,8 +19,6 @@ 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; @@ -42,6 +40,7 @@ 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; @@ -128,25 +127,25 @@ public void init(Context context) { @Override public SamzaSqlRelMessage apply(SamzaSqlInputMessage samzaSqlInputMessage) { - Instant startProcessing = Instant.now(); + long startProcessing = System.nanoTime(); /* 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.toString()); - updateMetrics(startProcessing, Instant.now()); + retMsg.setScanTime(startProcessing); + updateMetrics(startProcessing, System.nanoTime()); return retMsg; } /** * Updates the MetricsRegistery of this operator - * @param startProcessing = begin processing of the message - * @param endProcessing = end of processing + * @param startProcessing = begin (nanoTime) processing of the message + * @param endProcessing = end (nanoTime) of processing */ - private void updateMetrics(Instant startProcessing, Instant endProcessing) { + private void updateMetrics(long startProcessing, long endProcessing) { queryInputEvents.inc(); - processingTime.update(Duration.between(startProcessing, endProcessing).toMillis()); + processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing)); } } // 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 c2f9ea5f69..d788297c3b 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 = "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 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 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 new file mode 100644 index 0000000000..c9d0239611 --- /dev/null +++ b/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java @@ -0,0 +1,33 @@ +/* + * 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 dffb77e3a1..ac8e10dbe2 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,33 +36,34 @@ public class TestSamzaSqlRelMessage { @Test public void testGetField() { - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); 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("", "", "")); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, + new SamzaSqlRelMsgMetadata(0, 0, 0)); Assert.assertFalse(message.getSamzaSqlRelRecord().getField("field3").isPresent()); } @Test public void testEquality() { - SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); SamzaSqlRelMessage message2 = new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value1", "value2"), - new SamzaSqlRelMsgMetadata("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); Assert.assertEquals(message1, message2); Assert.assertEquals(message1.hashCode(), message2.hashCode()); } @Test public void testInEquality() { - SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); SamzaSqlRelMessage message2 = new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value2", "value2"), - new SamzaSqlRelMsgMetadata("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); Assert.assertNotEquals(message1, message2); Assert.assertNotEquals(message1.hashCode(), message2.hashCode()); } @@ -70,7 +71,7 @@ public void testInEquality() { @Test public void testCompositeKeyCreation() { List keyPartNames = Arrays.asList("kfield1", "kfield2"); - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); SamzaSqlRelRecord relRecord1 = SamzaSqlRelMessage.createSamzaSqlCompositeKey(message, Collections.singletonList(0)); Assert.assertEquals(relRecord1.getFieldNames().size(), 1); @@ -89,7 +90,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("", "", "")); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0)); 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 5afa4fe80c..689e8d2d29 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,7 +53,8 @@ public class TestSamzaSqlRelMessageSerde { @Test public void testWithDifferentFields() { - SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, + new SamzaSqlRelMsgMetadata(0, 0, 0)); 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 4e6ff4f8b9..6885fc1acf 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("", "", "")).getSamzaSqlRelRecord(); + new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0,0, 0)).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 2a98c8c9e9..be64cc4b84 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("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); 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 3a2b943f28..829532cb52 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("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); 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 53c7bc14cb..b92dfe8868 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,8 +44,10 @@ public class TestSamzaSqlLocalTableJoinFunction { @Test public void testWithInnerJoinWithTableOnRight() { - SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); - SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, + new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, + new SamzaSqlRelMsgMetadata(0, 0, 0)); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(0, 1); List tableKeyIds = Arrays.asList(0, 1); @@ -77,8 +79,10 @@ public void testWithInnerJoinWithTableOnRight() { @Test public void testWithInnerJoinWithTableOnLeft() { - SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); - SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", "")); + SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, + new SamzaSqlRelMsgMetadata(0, 0, 0)); + SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, + new SamzaSqlRelMsgMetadata(0, 0, 0)); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(0, 2); List tableKeyIds = Arrays.asList(0, 2); @@ -111,7 +115,7 @@ public void testWithInnerJoinWithTableOnLeft() { @Test public void testNullRecordWithInnerJoin() { SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(0, 1); List tableKeyIds = Arrays.asList(2, 3); @@ -135,7 +139,7 @@ public void testNullRecordWithInnerJoin() { @Test public void testNullRecordWithLeftOuterJoin() { SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, - new SamzaSqlRelMsgMetadata("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); 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 2deb6ad942..20cea14c30 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("", "", "")); + new SamzaSqlRelMsgMetadata(0, 0, 0)); SamzaSqlRelMessage tableMsg = relConverter.convertToRelMessage(new KV(tableRecord.get("id"), tableRecord)); JoinRelType joinRelType = JoinRelType.INNER; List streamKeyIds = Arrays.asList(1);