content: /**
   * 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.apex.malhar.kudu.sqltranslator;
  
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  
  import org.antlr.v4.runtime.tree.ParseTree;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.kudu.sqlparser.KuduSQLExpressionBaseListener;
  import org.apache.apex.malhar.kudu.sqlparser.KuduSQLExpressionParser;
  import org.apache.kudu.ColumnSchema;
  import org.apache.kudu.Common;
  import org.apache.kudu.client.KuduPredicate;
  
  import com.google.common.base.Preconditions;
  
  /**
   * This class builds Kudu equivalent scanner logic from a supplied SQL expression.
   * Note that the SQL expression is set in {@code KuduSQLParser}
   * The parser class uses a "tree walker" to walk the parsed tree and while doing so registers this class
   * This listener then start building the Kudu scanner constructs as it listens to the tree walk steps.
   * Note that it only only overrides required methods from its base class {@code KuduSQLExpressionBaseListener}
   * The KuduSQLExpressionBaseListener class is an autogenerated class by the Antlr4 compiler
   *
   * @since 3.8.0
   * */
  public class KuduSQLParseTreeListener extends KuduSQLExpressionBaseListener
  {
    private boolean isSuccessfullyParsed = true;
  
    private boolean isFilterExpressionEnabled = false;
  
    private boolean isOptionsEnabled = false;
  
    private Set<String> listOfColumnsUsed = new HashSet<>();
  
    private Map<String,String> aliases = new HashMap<>();
  
    private Map<String,String> optionsUsed = new HashMap<>();
  
    private List<KuduPredicate> kuduPredicateList = new ArrayList<>();
  
    private Map<String,ColumnSchema> columnSchemaLookups = new HashMap<>();
  
    private String tableName = null;
  
    private boolean isSelectStarExpressionEnabled = false;
  
    private static final transient Logger LOG = LoggerFactory.getLogger(KuduSQLParseTreeListener.class);
  
    private String controlTupleMessage = null;
  
    private Long readSnapshotTime = null;
  
    public static final String READ_SNAPSHOT_TIME = "read_snapshot_time";
  
    public static final String CONTROLTUPLE_MESSAGE = "controltuple_message";
  
    private void clearAll()
    {
      listOfColumnsUsed.clear();
      aliases.clear();
      optionsUsed.clear();
      kuduPredicateList.clear();
      tableName = null;
      columnSchemaLookups.clear();
    }
  
    public void setColumnSchemaList(List<ColumnSchema> listOfColumnsForCurrentTable)
    {
      Preconditions.checkNotNull(listOfColumnsForCurrentTable,"Column schemas " +
          "cannot be null for kudu table");
      for (ColumnSchema aColumnDef : listOfColumnsForCurrentTable) {
        columnSchemaLookups.put(aColumnDef.getName(),aColumnDef);
        aliases.put(aColumnDef.getName(),aColumnDef.getName()); // By default each column is its own alias in POJO.
      }
    }
  
    @Override
    public void exitKudusqlexpression(KuduSQLExpressionParser.KudusqlexpressionContext ctx)
    {
      LOG.debug(" Scanning " + listOfColumnsUsed.size() + " columns from " + tableName + " table");
      LOG.debug(" Number of predicates that are part of the scan are " + kuduPredicateList.size());
      LOG.debug(" Select * expression enabled " + isSelectStarExpressionEnabled);
      LOG.debug(" Filter expression enabled " + isFilterExpressionEnabled);
      LOG.debug(" Are options being enabled  " + isOptionsEnabled);
      LOG.debug(" Read snapshot time being used as " + readSnapshotTime);
      LOG.debug(" Control tuple message being used as " + controlTupleMessage);
    }
  
    @Override
    public void exitSELECT_ID_ONLY_USED_AS_COLUMN_NAME(KuduSQLExpressionParser.SELECT_ID_ONLY_USED_AS_COLUMN_NAMEContext
        ctx)
    {
      super.exitSELECT_ID_ONLY_USED_AS_COLUMN_NAME(ctx);
      KuduSQLExpressionParser.IdorcolumnnameContext ctxForIDOrColumnName = ctx.idorcolumnname();
      if (ctxForIDOrColumnName != null ) { // we can ignore complex expressions as they are resolved recursively
        // We are not inside as complex comma separated expression
        String columnName = extractColumnNameFromContext(ctxForIDOrColumnName);
        if ( columnName != null) {
          if (columnSchemaLookups.containsKey(columnName)) {
            LOG.debug(" Kudu column being enabled for scanning " + columnName );
            aliases.put(columnName,columnName);
            listOfColumnsUsed.add(columnName);
          } else {
            LOG.error("Invalid column  being set as a scanner ( column names are case sensitive ). " + columnName);
            isSuccessfullyParsed = false;
          }
        }
      }
    }
  
  
    @Override
    public void exitALL_COLUMNS_SELECT_EXP(KuduSQLExpressionParser.ALL_COLUMNS_SELECT_EXPContext ctx)
    {
      super.exitALL_COLUMNS_SELECT_EXP(ctx);
      listOfColumnsUsed.addAll(columnSchemaLookups.keySet());
      isSelectStarExpressionEnabled = true;
    }
  
    private String extractColumnNameFromContext(KuduSQLExpressionParser.IdorcolumnnameContext ctxForIDOrColumnName)
    {
      if (ctxForIDOrColumnName.ID() != null) {
        // we are dealing with a non-reserved keyword as a columnname.
        return ctxForIDOrColumnName.ID().getSymbol().getText();
      } else {
        for (int i = 0; i < ctxForIDOrColumnName.getChildCount(); i++) {
          // iterate all the terminal node children to see wihich keyword is used. Note that there are spaces possible
          ParseTree terminalNodeTree = ctxForIDOrColumnName.getChild(i);
          String childNodeText = terminalNodeTree.getText();
          if ( (!childNodeText.equalsIgnoreCase(" ")) &&
              (!childNodeText.equalsIgnoreCase("'"))
              ) {
            return childNodeText; // Anything other thant a quote or whitespace is the column name ( reserved )
          }
        } // end for loop for terminal node children
      } // conditional for non-ID column names
      return null;
    }
  
    @Override
    public void exitSELECT_ALIAS_USED(KuduSQLExpressionParser.SELECT_ALIAS_USEDContext ctx)
    {
      super.exitSELECT_ALIAS_USED(ctx);
      KuduSQLExpressionParser.IdorcolumnnameContext ctxForIDOrColumnName = ctx.idorcolumnname();
      if (ctxForIDOrColumnName != null ) { // we can ignore complex expressions as they are resolved recursively
        // We are not inside as complex comma separated expression
        String columnName = extractColumnNameFromContext(ctxForIDOrColumnName);
        if ( columnName != null) {
          if (columnSchemaLookups.containsKey(columnName)) {
            LOG.debug(" Kudu column being enabled for scanning " + columnName );
            LOG.debug(columnName + " is being scanned as " + ctx.ID().getSymbol().getText());
            aliases.put(columnName,ctx.ID().getSymbol().getText());
            listOfColumnsUsed.add(columnName);
          } else {
            LOG.error("Invalid column  being set as a scanner ( column names are case sensitive ). " + columnName);
            isSuccessfullyParsed = false;
          }
        }
      } // end null check only for id or column names
    }
  
    @Override
    public void exitTableclause(KuduSQLExpressionParser.TableclauseContext ctx)
    {
      super.exitTableclause(ctx);
      tableName = ctx.ID().getText();
    }
  
    @Override
    public void exitWhereclause(KuduSQLExpressionParser.WhereclauseContext ctx)
    {
      super.exitWhereclause(ctx);
      isFilterExpressionEnabled = true;
    }
  
    @Override
    public void exitWithoptionsclause(KuduSQLExpressionParser.WithoptionsclauseContext ctx)
    {
      super.exitWithoptionsclause(ctx);
      isOptionsEnabled = true;
    }
  
    private KuduPredicate buildKuduPredicate(String columnName,KuduPredicate.ComparisonOp comparisonOp,
        KuduSQLExpressionParser.AnyvalueContext anyvalueContext)
    {
      ColumnSchema thisColumnSchema = columnSchemaLookups.get(columnName);
      if (anyvalueContext.bool() != null) {
        if (thisColumnSchema.getType().getDataType().getNumber() != (Common.DataType.BOOL_VALUE)) {
          LOG.error(" Mismatched data type for column " + columnName);
          isSuccessfullyParsed  = false;
          return null;
        }
        return KuduPredicate.newComparisonPredicate(thisColumnSchema,comparisonOp,
          Boolean.valueOf(anyvalueContext.bool().getText().toLowerCase()));
      }
      if (anyvalueContext.doubleval() != null) {
        if (thisColumnSchema.getType().getDataType().getNumber() != (Common.DataType.DOUBLE_VALUE)) {
          LOG.error(" Mismatched data type for column (Ensure doubles are appended with letter d)" + columnName);
          isSuccessfullyParsed  = false;
          return null;
        }
        return KuduPredicate.newComparisonPredicate(thisColumnSchema,comparisonOp,
          Double.valueOf(anyvalueContext.doubleval().getText()));
      }
      if (anyvalueContext.floatval() != null) {
        if (thisColumnSchema.getType().getDataType().getNumber() != (Common.DataType.FLOAT_VALUE)) {
          LOG.error(" Mismatched data type for column (Ensure doubles are appended with letter f)" + columnName);
          isSuccessfullyParsed  = false;
          return null;
        }
        return KuduPredicate.newComparisonPredicate(thisColumnSchema,comparisonOp,
          Float.valueOf(anyvalueContext.floatval().getText()));
      }
      if (anyvalueContext.stringval() != null) {
        if ( (thisColumnSchema.getType().getDataType().getNumber() != (Common.DataType.STRING_VALUE)) &&
            (thisColumnSchema.getType().getDataType().getNumber() != (Common.DataType.BINARY_VALUE))) {
          LOG.error(" Mismatched data type for column ( Has to be a string or a binary value enclosed in" +
              " double quotes" + columnName);
          isSuccessfullyParsed  = false;
          return null;
        }
        if (thisColumnSchema.getType().getDataType().getNumber() == (Common.DataType.STRING_VALUE)) {
          return KuduPredicate.newComparisonPredicate(thisColumnSchema, comparisonOp,
            anyvalueContext.stringval().getText());
        }
        if (thisColumnSchema.getType().getDataType().getNumber() == (Common.DataType.BINARY_VALUE)) {
          return KuduPredicate.newComparisonPredicate(thisColumnSchema, comparisonOp,
            anyvalueContext.stringval().getText().getBytes());
        }
      }
      if (anyvalueContext.numval() != null) {
        int dataTypeNumberForKuduCol = thisColumnSchema.getType().getDataType().getNumber();
        if (
            (dataTypeNumberForKuduCol != (Common.DataType.UNIXTIME_MICROS_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.INT8_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.INT16_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.INT32_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.INT64_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.UINT8_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.UINT16_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.UINT32_VALUE)) &&
            (dataTypeNumberForKuduCol != (Common.DataType.UINT64_VALUE))
            ) {
          LOG.error(" Mismatched data type for column " + columnName);
          isSuccessfullyParsed  = false;
          return null;
        }
        return KuduPredicate.newComparisonPredicate(thisColumnSchema,comparisonOp,
          Long.valueOf(anyvalueContext.numval().getText()));
      }
      return null;
    }
  
    @Override
    public void exitFILTER_COMPARISION_EXP(KuduSQLExpressionParser.FILTER_COMPARISION_EXPContext ctx)
    {
      super.exitFILTER_COMPARISION_EXP(ctx);
      String columnName = extractColumnNameFromContext(ctx.idorcolumnname());
      if (columnSchemaLookups.containsKey(columnName)) {
        if (ctx.comparisionoperator().EQUAL_TO() != null) {
          kuduPredicateList.add(buildKuduPredicate(columnName,KuduPredicate.ComparisonOp.EQUAL,ctx.anyvalue()));
        }
        if (ctx.comparisionoperator().GREATER_THAN() != null) {
          kuduPredicateList.add(buildKuduPredicate(columnName,KuduPredicate.ComparisonOp.GREATER,ctx.anyvalue()));
        }
        if (ctx.comparisionoperator().LESSER_THAN() != null) {
          kuduPredicateList.add(buildKuduPredicate(columnName,KuduPredicate.ComparisonOp.LESS,ctx.anyvalue()));
        }
        if (ctx.comparisionoperator().GREATER_THAN_OR_EQUAL() != null) {
          kuduPredicateList.add(buildKuduPredicate(columnName,KuduPredicate.ComparisonOp.GREATER_EQUAL,ctx.anyvalue()));
        }
        if (ctx.comparisionoperator().LESSER_THAN_OR_EQUAL() != null) {
          kuduPredicateList.add(buildKuduPredicate(columnName,KuduPredicate.ComparisonOp.LESS_EQUAL,ctx.anyvalue()));
        }
      } else {
        LOG.error(columnName + " is not a valid column name for this kudu table");
        isSuccessfullyParsed = false;
      }
    }
  
    @Override
    public void exitIS_NULL_FILTER_EXP(KuduSQLExpressionParser.IS_NULL_FILTER_EXPContext ctx)
    {
      super.exitIS_NULL_FILTER_EXP(ctx);
      String columnName = extractColumnNameFromContext(ctx.idorcolumnname());
      if (columnSchemaLookups.containsKey(columnName)) {
        kuduPredicateList.add(KuduPredicate.newIsNullPredicate(columnSchemaLookups.get(columnName)));
      } else {
        LOG.error(columnName + " is not a valid column name for this kudu table");
        isSuccessfullyParsed = false;
      }
  
    }
  
    private List<Boolean> extractListOfBools(KuduSQLExpressionParser.ListofboolsContext listofboolsContext,
        String columnName)
    {
      List<Boolean> returnList = new ArrayList<>();
      for (KuduSQLExpressionParser.BoolContext aBool: listofboolsContext.bool()) {
        returnList.add(Boolean.valueOf(aBool.getText().toLowerCase()));
      }
      if (returnList.size() == 0) {
        LOG.error("Empty list of booleans specified for IN expression for column filter " + columnName);
        isSuccessfullyParsed = false;
      }
      return returnList;
    }
  
  
    private List<Double> extractListOfDoubles(KuduSQLExpressionParser.ListofdoublesContext listofdoubles,
        String columnName)
    {
      List<Double> returnList = new ArrayList<>();
      for (KuduSQLExpressionParser.DoublevalContext aDouble: listofdoubles.doubleval()) {
        returnList.add(Double.valueOf(aDouble.getText()));
      }
      if (returnList.size() == 0) {
        LOG.error("Empty list of doubles specified for IN expression for column filter " + columnName);
        isSuccessfullyParsed = false;
      }
      return returnList;
    }
  
    private List<Float> extractListOfFloats(KuduSQLExpressionParser.ListoffloatsContext listoffloatsContext,
        String columnName)
    {
      List<Float> returnList = new ArrayList<>();
      for (KuduSQLExpressionParser.FloatvalContext aFloat: listoffloatsContext.floatval()) {
        returnList.add(Float.valueOf(aFloat.getText()));
      }
      if (returnList.size() == 0) {
        LOG.error("Empty list of floats specified for IN expression for column filter " + columnName);
        isSuccessfullyParsed = false;
      }
      return returnList;
    }
  
  
    private List<Long> extractListOfLongs(KuduSQLExpressionParser.ListofnumsContext listofnumsContext,
        String columnName)
    {
      List<Long> returnList = new ArrayList<>();
      for (KuduSQLExpressionParser.NumvalContext aLong: listofnumsContext.numval()) {
        returnList.add(Long.valueOf(aLong.getText()));
      }
      if (returnList.size() == 0) {
        LOG.error("Empty list of longs specified for IN expression for column filter " + columnName);
        isSuccessfullyParsed = false;
      }
      return returnList;
    }
  
  
    private List<String> extractListOfStrings(KuduSQLExpressionParser.ListofstringsContext listofstringsContext,
        String columnName)
    {
      List<String> returnList = new ArrayList<>();
      for (KuduSQLExpressionParser.StringvalContext aString: listofstringsContext.stringval()) {
        returnList.add(aString.getText());
      }
      if (returnList.size() == 0) {
        LOG.error("Empty list of Strings specified for IN expression for column filter " + columnName);
        isSuccessfullyParsed = false;
      }
      return returnList;
    }
  
  
    private List<byte[]> extractListOfBinary(KuduSQLExpressionParser.ListofstringsContext listofstringsContext,
        String columnName)
    {
  
      List<byte[]> returnList = new ArrayList<>();
      for (KuduSQLExpressionParser.StringvalContext aString: listofstringsContext.stringval()) {
        returnList.add(aString.getText().getBytes());
      }
      if (returnList.size() == 0) {
        LOG.error("Empty list of binaries specified for IN expression for column filter " + columnName);
        isSuccessfullyParsed = false;
      }
      return returnList;
    }
  
  
  
    @Override
    public void exitIN_FILTER_EXP(KuduSQLExpressionParser.IN_FILTER_EXPContext ctx)
    {
      super.exitIN_FILTER_EXP(ctx);
      String columnName = extractColumnNameFromContext(ctx.idorcolumnname());
      if (columnSchemaLookups.containsKey(columnName)) {
        KuduSQLParser.ListofanyvalueContext  listofanyvalueContext = ctx.listofanyvalue();
        // sort out an IN predicate for a list of bools
        KuduSQLExpressionParser.ListofboolsContext listofboolsContext = listofanyvalueContext.listofbools();
        if (listofboolsContext != null) {
          kuduPredicateList.add(KuduPredicate.newInListPredicate(columnSchemaLookups.get(columnName),
              extractListOfBools(listofboolsContext,columnName)));
        }
  
        // sort out an IN predicate for a list of doubles
        KuduSQLExpressionParser.ListofdoublesContext listofdoubles = listofanyvalueContext.listofdoubles();
        if (listofdoubles != null) {
          kuduPredicateList.add(KuduPredicate.newInListPredicate(columnSchemaLookups.get(columnName),
              extractListOfDoubles(listofdoubles,columnName)));
        }
  
        // sort out an IN predicate for a list of floats
        KuduSQLExpressionParser.ListoffloatsContext listoffloatsContext = listofanyvalueContext.listoffloats();
        if (listoffloatsContext != null) {
          kuduPredicateList.add(KuduPredicate.newInListPredicate(columnSchemaLookups.get(columnName),
              extractListOfFloats(listoffloatsContext,columnName)));
        }
  
        // sort out an IN predicate for a list of longs
        KuduSQLExpressionParser.ListofnumsContext listofnumsContext = listofanyvalueContext.listofnums();
        if (listofnumsContext != null) {
          kuduPredicateList.add(KuduPredicate.newInListPredicate(columnSchemaLookups.get(columnName),
              extractListOfLongs(listofnumsContext,columnName)));
        }
  
        // sort out an IN predicate for a list of strings or binary
        KuduSQLExpressionParser.ListofstringsContext listofstringsContext = listofanyvalueContext.listofstrings();
        if (listofstringsContext != null) {
          if (columnSchemaLookups.get(columnName).getType().getDataType().getNumber() == Common.DataType.STRING_VALUE) {
            kuduPredicateList.add(KuduPredicate.newInListPredicate(columnSchemaLookups.get(columnName),
                extractListOfStrings(listofstringsContext,columnName)));
          }
          if (columnSchemaLookups.get(columnName).getType().getDataType().getNumber() == Common.DataType.STRING_VALUE) {
            kuduPredicateList.add(KuduPredicate.newInListPredicate(columnSchemaLookups.get(columnName),
                extractListOfBinary(listofstringsContext,columnName)));
          }
        }
  
      } else {
        LOG.error(columnName + " is not a valid column name to be used in a where clause for this kudu table");
        isSuccessfullyParsed = false;
      }
    }
  
    @Override
    public void exitIS_NOT_NULL_FILTER_EXP(KuduSQLExpressionParser.IS_NOT_NULL_FILTER_EXPContext ctx)
    {
      super.exitIS_NOT_NULL_FILTER_EXP(ctx);
      String columnName = extractColumnNameFromContext(ctx.idorcolumnname());
      if (columnSchemaLookups.containsKey(columnName)) {
        kuduPredicateList.add(KuduPredicate.newIsNotNullPredicate(columnSchemaLookups.get(columnName)));
      } else {
        LOG.error(columnName + " is not a valid column name for this kudu table");
        isSuccessfullyParsed = false;
      }
    }
  
    @Override
    public void exitSET_CONTROL_TUPLE_MSG(KuduSQLExpressionParser.SET_CONTROL_TUPLE_MSGContext ctx)
    {
      super.exitSET_CONTROL_TUPLE_MSG(ctx);
      controlTupleMessage = ctx.STRINGVAL().getText();
      optionsUsed.put(CONTROLTUPLE_MESSAGE,controlTupleMessage);
    }
  
    @Override
    public void exitSET_READ_SNAPSHOT_TIME(KuduSQLExpressionParser.SET_READ_SNAPSHOT_TIMEContext ctx)
    {
      super.exitSET_READ_SNAPSHOT_TIME(ctx);
      readSnapshotTime = Long.valueOf(ctx.INT().getText());
      optionsUsed.put(READ_SNAPSHOT_TIME, "" + readSnapshotTime);
    }
  
    public boolean isSuccessfullyParsed()
    {
      return isSuccessfullyParsed;
    }
  
    public void setSuccessfullyParsed(boolean successfullyParsed)
    {
      isSuccessfullyParsed = successfullyParsed;
    }
  
    public Set<String> getListOfColumnsUsed()
    {
      return listOfColumnsUsed;
    }
  
    public void setListOfColumnsUsed(Set<String> listOfColumnsUsed)
    {
      this.listOfColumnsUsed = listOfColumnsUsed;
    }
  
    public Map<String, String> getAliases()
    {
      return aliases;
    }
  
    public void setAliases(Map<String, String> aliases)
    {
      this.aliases = aliases;
    }
  
    public Map<String, String> getOptionsUsed()
    {
      return optionsUsed;
    }
  
    public void setOptionsUsed(Map<String, String> optionsUsed)
    {
      this.optionsUsed = optionsUsed;
    }
  
    public List<KuduPredicate> getKuduPredicateList()
    {
      return kuduPredicateList;
    }
  
    public void setKuduPredicateList(List<KuduPredicate> kuduPredicateList)
    {
      this.kuduPredicateList = kuduPredicateList;
    }
  
    public String getTableName()
    {
      return tableName;
    }
  
    public void setTableName(String tableName)
    {
      this.tableName = tableName;
    }
  
    public boolean isSelectStarExpressionEnabled()
    {
      return isSelectStarExpressionEnabled;
    }
  
    public void setSelectStarExpressionEnabled(boolean selectStarExpressionEnabled)
    {
      isSelectStarExpressionEnabled = selectStarExpressionEnabled;
    }
  
    public boolean isFilterExpressionEnabled()
    {
      return isFilterExpressionEnabled;
    }
  
    public void setFilterExpressionEnabled(boolean filterExpressionEnabled)
    {
      isFilterExpressionEnabled = filterExpressionEnabled;
    }
  
    public boolean isOptionsEnabled()
    {
      return isOptionsEnabled;
    }
  
    public void setOptionsEnabled(boolean optionsEnabled)
    {
      isOptionsEnabled = optionsEnabled;
    }
  
    public String getControlTupleMessage()
    {
      return controlTupleMessage;
    }
  
    public void setControlTupleMessage(String controlTupleMessage)
    {
      this.controlTupleMessage = controlTupleMessage;
    }
  
    public Long getReadSnapshotTime()
    {
      return readSnapshotTime;
    }
  
    public void setReadSnapshotTime(Long readSnapshotTime)
    {
      this.readSnapshotTime = readSnapshotTime;
    }
  }
  