Merge remote-tracking branch 'origin/master' into external-hive-support

cluster/src/test/scala/org/apache/spark/sql/store/BugTest.scala

@@ -422,4 +422,58 @@ class BugTest extends SnappyFunSuite with BeforeAndAfterAll {
     conn.close()
     TestUtil.stopNetServer()
   }
+
+  test("Bug SNAP-2758 . view containing aggregate function & join throws error") {
+    snc
+    var serverHostPort2 = TestUtil.startNetServer()
+    var conn = DriverManager.getConnection(s"jdbc:snappydata://$serverHostPort2")
+    var stmt = conn.createStatement()
+    val snappy = snc.snappySession
+    snappy.sql("drop table if exists test1")
+    snappy.sql("create table test1 (col1_1 int, col1_2 int, col1_3 int, col1_4 string) " +
+        "using column ")
+
+    snappy.sql("create table test2 (col2_1 int, col2_2 int,  col2_3 int, col2_5 string) " +
+        "using column ")
+
+    snappy.sql(" CREATE OR REPLACE VIEW v1 as select col2_1, col2_2, " +
+        "col2_5 as longtext from test2 where col2_3 > 10")
+
+    val q1 = "select a.col1_1, a.col1_2, " +
+        " CASE WHEN a.col1_4 = '' THEN '#' ELSE a.col1_4 END functionalAreaCode," +
+        "b.longtext as name, " +
+        " sum(a.col1_3)" +
+        "from test1 a left outer join v1 as b on a.col1_1 = b.col2_1" +
+        " group by a.col1_1, a.col1_2, " +
+        " CASE WHEN a.col1_4  = '' THEN '#' ELSE a.col1_4  END," +
+        " b.longtext "
+    snappy.sql(q1)
+    snappy.sql(s" CREATE OR REPLACE VIEW v3 as $q1")
+
+    val q = "select a.col1_1, a.col1_2, " +
+        " CASE WHEN a.col1_4 = '' THEN '#' ELSE a.col1_4 END functionalAreaCode," +
+        "'#' as fsid,  " +
+        "b.longtext as name, " +
+        " sum(a.col1_3)" +
+        "from test1 a left outer join v1 as b on a.col1_1 = b.col2_1" +
+        " group by a.col1_1, a.col1_2, " +
+        " CASE WHEN a.col1_4  = '' THEN '#' ELSE a.col1_4  END," +
+        " '#'," +
+        " b.longtext "
+    snappy.sql(q)
+    snappy.sql(s" CREATE OR REPLACE VIEW v2 as $q")
+    snappy.sql("select count(*) from v2").collect()
+
+    stmt.execute("drop view v3")
+    stmt.execute("drop view v2")
+    stmt.execute("drop view v1")
+    snc.sql("drop table if exists test1")
+    snc.sql("drop table if exists test2")
+
+    conn.close()
+    TestUtil.stopNetServer()
+
+  }
+
+
 }

core/src/main/scala/io/snappydata/Literals.scala

@@ -29,6 +29,7 @@ object StreamingConstants {
   val STREAM_QUERY_ID = "streamqueryid"
   val SINK_CALLBACK = "sinkcallback"
   val CONFLATION = "conflation"
+  val EVENT_COUNT_COLUMN = "SNAPPYSYS_INTERNAL____EVENT_COUNT"
 
   object EventType {
     val INSERT = 0

core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParamLiteral.scala

@@ -355,6 +355,8 @@ case class ParamLiteral(var value: Any, var dataType: DataType,
 final class RefParamLiteral(val param: ParamLiteral, _value: Any, _dataType: DataType, _pos: Int)
     extends ParamLiteral(_value, _dataType, _pos, execId = param.execId) {
 
+  assert(!param.isInstanceOf[RefParamLiteral])
+
   private[sql] def referenceEquals(p: ParamLiteral): Boolean = {
     if (param eq p) {
       // Check that value and dataType should also be equal at this point.
@@ -423,57 +425,67 @@ trait ParamLiteralHolder {
 
   private[sql] final def getCurrentParamsId: Int = paramListId
 
+  /**
+   * Find existing ParamLiteral with given value and DataType. This should
+   * never return a RefParamLiteral.
+   */
   private def findExistingParamLiteral(value: Any, dataType: DataType,
-      numConstants: Int): ParamLiteral = {
+      numConstants: Int): Option[ParamLiteral] = {
     // for size >= 4 use a lookup map to search for same constant else linear search
     if (numConstants >= 4) {
       if (paramConstantMap eq null) {
         // populate the map while checking for a match
         paramConstantMap = UnifiedMap.newMap(8)
         var i = 0
-        var existing: ParamLiteral = null
+        var existing: Option[ParamLiteral] = None
         while (i < numConstants) {
-          val param = parameterizedConstants(i)
-          if ((existing eq null) && dataType == param.dataType && value == param.value) {
-            existing = param
+          parameterizedConstants(i) match {
+            case _: RefParamLiteral => // skip
+            case param =>
+              if (existing.isEmpty && dataType == param.dataType && value == param.value) {
+                existing = Some(param)
+              }
+              paramConstantMap.put(param.dataType -> param.value, param)
           }
-          paramConstantMap.put(param.dataType -> param.value, param)
           i += 1
         }
         existing
-      } else paramConstantMap.get(dataType -> value)
+      } else Option(paramConstantMap.get(dataType -> value))
     } else {
       var i = 0
       while (i < numConstants) {
-        val param = parameterizedConstants(i)
-        if (dataType == param.dataType && value == param.value) {
-          return param
+        parameterizedConstants(i) match {
+          case _: RefParamLiteral => // skip
+          case param =>
+            if (dataType == param.dataType && value == param.value) {
+              return Some(param)
+            }
         }
         i += 1
       }
-      null
+      None
     }
   }
 
   private[sql] final def addParamLiteralToContext(value: Any,
       dataType: DataType): ParamLiteral = {
     val numConstants = parameterizedConstants.length
-    val existing = findExistingParamLiteral(value, dataType, numConstants)
-    if (existing ne null) {
-      // Add to paramelizedConstants list so that its position can be updated
-      // if required (e.g. if a ParamLiteral is reverted to a Literal for
-      //   functions that require so as in SnappyParserConsts.FOLDABLE_FUNCTIONS)
-      // In addition RefParamLiteral maintains its own copy of value to avoid updating
-      // the referenced ParamLiteral's value by functions like ROUND, so that needs to
-      // be changed too when a plan with updated tokens is created.
-      val ref = new RefParamLiteral(existing, value, dataType, numConstants)
-      parameterizedConstants += ref
-      ref
-    } else {
-      val p = ParamLiteral(value, dataType, numConstants, paramListId)
-      parameterizedConstants += p
-      if (paramConstantMap ne null) paramConstantMap.put(dataType -> value, p)
-      p
+    findExistingParamLiteral(value, dataType, numConstants) match {
+      case None =>
+        val p = ParamLiteral(value, dataType, numConstants, paramListId)
+        parameterizedConstants += p
+        if (paramConstantMap ne null) paramConstantMap.put(dataType -> value, p)
+        p
+      case Some(existing) =>
+        // Add to parameterizedConstants list so that its position can be updated
+        // if required (e.g. if a ParamLiteral is reverted to a Literal for
+        //   functions that require so as in SnappyParserConsts.FOLDABLE_FUNCTIONS)
+        // In addition RefParamLiteral maintains its own copy of value to avoid updating
+        // the referenced ParamLiteral's value by functions like ROUND, so that needs to
+        // be changed too when a plan with updated tokens is created.
+        val ref = new RefParamLiteral(existing, value, dataType, numConstants)
+        parameterizedConstants += ref
+        ref
     }
   }
 

core/src/main/scala/org/apache/spark/sql/streaming/SnappySinkCallback.scala

@@ -108,15 +108,15 @@ case class SnappyStoreSink(snappySession: SnappySession,
       }
     }
 
-    val hashAggregateSizeChanged = HashAggregateSize.get(snappySession.sessionState.conf)
+    val hashAggregateSizeIsDefault = HashAggregateSize.get(snappySession.sessionState.conf)
         .equals(HashAggregateSize.defaultValue.get)
-    if (hashAggregateSizeChanged) {
+    if (hashAggregateSizeIsDefault) {
       HashAggregateSize.set(snappySession.sessionState.conf, "10m")
     }
     try {
       sinkCallback.process(snappySession, parameters, batchId, convert(data), posDup)
     } finally {
-      if (hashAggregateSizeChanged) {
+      if (hashAggregateSizeIsDefault) {
         HashAggregateSize.set(snappySession.sessionState.conf, HashAggregateSize.defaultValue.get)
       }
     }
@@ -148,7 +148,6 @@ import org.apache.spark.sql.snappy._
 class DefaultSnappySinkCallback extends SnappySinkCallback {
   def process(snappySession: SnappySession, parameters: Map[String, String],
       batchId: Long, df: Dataset[Row], posDup: Boolean) {
-    df.cache().count()
     log.debug(s"Processing batchId $batchId with parameters $parameters ...")
     val tableName = snappySession.sessionCatalog.formatTableName(parameters(TABLE_NAME))
     val conflationEnabled = if (parameters.contains(CONFLATION)) {
@@ -163,6 +162,7 @@ class DefaultSnappySinkCallback extends SnappySinkCallback {
         s", eventTypeColumnAvailable:$eventTypeColumnAvailable,possible duplicate: $posDup")
 
     if (keyColumns.nonEmpty) {
+      df.cache().count()
       val dataFrame: DataFrame = if (conflationEnabled) getConflatedDf else df
       if (eventTypeColumnAvailable) {
         val deleteDf = dataFrame.filter(dataFrame(EVENT_TYPE_COLUMN) === EventType.DELETE)
@@ -202,7 +202,6 @@ class DefaultSnappySinkCallback extends SnappySinkCallback {
 
     log.debug(s"Processing batchId $batchId with parameters $parameters ... Done.")
 
-
     // We are grouping by key columns and getting the last record.
     // Note that this approach will work as far as the incoming dataframe is partitioned
     // by key columns and events are available in the correct order in the respective partition.
@@ -218,12 +217,20 @@ class DefaultSnappySinkCallback extends SnappySinkCallback {
         index += 1
         contains
       }
-      val conflatedDf = if (otherCols.isEmpty) df.distinct()
-      else {
-        val exprs = otherCols.map(c => last(c).alias(c))
+
+      val conflatedDf: DataFrame = {
+        val exprs = otherCols.map(c => last(c).alias(c)) ++
+            Seq(count(lit(1)).alias(EVENT_COUNT_COLUMN))
+
+        // if event type of the last event for a key is insert and there are more than one
+        // events for the same key, then convert inserts to put into
+        val columns = df.columns.filter(_ != EVENT_TYPE_COLUMN).map(col) ++
+            Seq(when(col(EVENT_TYPE_COLUMN) === EventType.INSERT && col(EVENT_COUNT_COLUMN) > 1,
+              EventType.UPDATE).otherwise(col(EVENT_TYPE_COLUMN)).alias(EVENT_TYPE_COLUMN))
+
         df.groupBy(keyCols.head, keyCols.tail: _*)
             .agg(exprs.head, exprs.tail: _*)
-            .select(df.columns.head, df.columns.tail: _*)
+            .select(columns: _*)
       }
       conflatedDf.cache()
     }

core/src/test/scala/org/apache/spark/sql/streaming/SnappyStoreSinkProviderSuite.scala

@@ -276,6 +276,24 @@ class SnappyStoreSinkProviderSuite extends SnappyFunSuite
     assertData(Array(Row(1, "name999", 999, "lname1")))
   }
 
+  test("[SNAP-2745]-conflation: delete,insert") {
+    val testId = testIdGenerator.getAndIncrement()
+    createTable()()
+    val topic = getTopic(testId)
+    kafkaTestUtils.createTopic(topic, partitions = 1)
+
+    val batch1 = Seq(Seq(1, "name1", 30, "lname1", 0))
+    kafkaTestUtils.sendMessages(topic, batch1.map(r => r.mkString(",")).toArray)
+    val streamingQuery = createAndStartStreamingQuery(topic, testId, conflation = true)
+
+    waitTillTheBatchIsPickedForProcessing(0, testId)
+    val batch2 = Seq(Seq(1, "name1", 30, "lname1", 2), Seq(1, "name1", 30, "lname1", 0))
+    kafkaTestUtils.sendMessages(topic, batch2.map(r => r.mkString(",")).toArray)
+
+    streamingQuery.processAllAvailable()
+
+    assertData(Array(Row(1, "name1", 30, "lname1")))
+  }
 
   test("test conflation disabled") {
     val testId = testIdGenerator.getAndIncrement()