Indexing tasks containing thetaSketches results in incorrect sketch values

[a2l007]

Affected Version

0.13.0-incubating

Description

One of our clusters running on 0.13.0 is seeing data corruption issues with thetasketches. There are two datasources used in this testcase: basefact and slice. slice is essentially same as basefact except with fewer dimensions, so as to improve the rollup ratio.
Ingestion Spec for basefact datasource:

{
  "type" : "index_hadoop",
  "id" : "index_base",
  "spec" : {
    "dataSchema" : {
      "dataSource" : "basefact",
      "parser" : {
        "type" : "avro_hadoop",
        "parseSpec" : {
          "format" : "avro",
          "timestampSpec" : {
            "column" : "date",
            "format" : "yyyyMMdd"
          },
          "dimensionsSpec" : {
            "dimensions" : [ "src_pty_id" ]
          }
        }
      },
      "metricsSpec" : [ {
        "type" : "thetaSketch",
        "name" : "test_sketch",
        "fieldName" : "test_sketch",
        "size" : 1048576,
        "shouldFinalize" : true,
        "isInputThetaSketch" : true,
        "errorBoundsStdDev" : null
      } ],
      "granularitySpec" : {
        "type" : "uniform",
        "segmentGranularity" : "DAY",
        "queryGranularity" : {
          "type" : "none"
        },
        "rollup" : true,
        "intervals" : [ "2019-05-22T00:00:00.000Z/2019-05-23T00:00:00.000Z" ]
      },
      "transformSpec" : {
        "filter" : null,
        "transforms" : [ ]
      }
    },
    "ioConfig" : {
      "type" : "hadoop",
      "inputSpec" : {
        "type" : "static",
        "paths" : "///projects/indexData",
        "inputFormat" : "org.apache.druid.data.input.avro.AvroValueInputFormat"
      },
      "metadataUpdateSpec" : null,
      "segmentOutputPath" : null
    },
    "tuningConfig" : {
      "type" : "hadoop",
      "workingPath" : null,
      "version" : "2019-05-23T08:33:19.990Z",
      "partitionsSpec" : {
        "type" : "hashed",
        "targetPartitionSize" : 1050000,
        "maxPartitionSize" : 1575000,
        "assumeGrouped" : true,
        "numShards" : -1,
        "partitionDimensions" : [ ]
      },
      "shardSpecs" : { },
      "indexSpec" : {
        "bitmap" : {
          "type" : "concise"
        },
        "dimensionCompression" : "lz4",
        "metricCompression" : "lz4",
        "longEncoding" : "longs"
      },
      "maxRowsInMemory" : 150000,
      "maxBytesInMemory" : -1,
      "leaveIntermediate" : false,
      "cleanupOnFailure" : true,
      "overwriteFiles" : false,
      "ignoreInvalidRows" : false,
      "jobProperties" : {
        "fs.permissions.umask-mode" : "027"
      },
      "combineText" : false,
      "useCombiner" : false,
      "buildV9Directly" : true,
      "numBackgroundPersistThreads" : 0,
      "forceExtendableShardSpecs" : false,
      "useExplicitVersion" : false,
      "allowedHadoopPrefix" : [ ],
      "logParseExceptions" : false,
      "maxParseExceptions" : 0
    }
  },
  "hadoopDependencyCoordinates" : null,
  "classpathPrefix" : null,
  "context" : { }
}

Ingestion spec for slice datasource:

{
  "type" : "index_hadoop",
  "id" : "index_slice",
  "spec" : {
    "dataSchema" : {
      "dataSource" : "slice",
      "parser" : {
        "type" : "avro_hadoop",
        "parseSpec" : {
          "format" : "avro",
          "timestampSpec" : {
            "column" : "date",
            "format" : "yyyyMMdd"
          },
          "dimensionsSpec" : {
            "dimensions" : [ "src_pty_id" ]
          }
        }
      },
      "metricsSpec" : [ {
      
        "type" : "thetaSketch",
        "name" : "test_sketch",
        "fieldName" : "test_sketch",
        "size" : 131072,
        "shouldFinalize" : true,
        "isInputThetaSketch" : true,
        "errorBoundsStdDev" : null
      } ],
      "granularitySpec" : {
        "type" : "uniform",
        "segmentGranularity" : "DAY",
        "queryGranularity" : {
          "type" : "none"
        },
        "rollup" : true,
        "intervals" : [ "2019-05-22T00:00:00.000Z/2019-05-23T00:00:00.000Z" ]
      },
      "transformSpec" : {
        "filter" : null,
        "transforms" : [ ]
      }
    },
    "ioConfig" : {
      "type" : "hadoop",
      "inputSpec" : {
        "type" : "dataSource",
        "ingestionSpec" : {
          "dataSource" : "basefact",
          "intervals" : [ "2019-05-22T00:00:00Z/P1D" ]
        }
      },
      "metadataUpdateSpec" : null,
      "segmentOutputPath" : null
    },
    "tuningConfig" : {
      "type" : "hadoop",
      "workingPath" : null,
      "version" : "2019-05-23T08:54:27.227Z",
      "partitionsSpec" : {
        "type" : "hashed",
        "targetPartitionSize" : 133000,
        "maxPartitionSize" : 199500,
        "assumeGrouped" : true,
        "numShards" : -1,
        "partitionDimensions" : [ ]
      },
      "shardSpecs" : { },
      "indexSpec" : {
        "bitmap" : {
          "type" : "concise"
        },
        "dimensionCompression" : "lz4",
        "metricCompression" : "lz4",
        "longEncoding" : "longs"
      },
      "maxRowsInMemory" : 10000,
      "maxBytesInMemory" : -1,
      "leaveIntermediate" : false,
      "cleanupOnFailure" : true,
      "overwriteFiles" : false,
      "ignoreInvalidRows" : false,
      "jobProperties" : {
        "fs.permissions.umask-mode" : "027"
      },
      "combineText" : false,
      "useCombiner" : false,
      "buildV9Directly" : true,
      "numBackgroundPersistThreads" : 0,
      "forceExtendableShardSpecs" : false,
      "useExplicitVersion" : false,
      "allowedHadoopPrefix" : [ ],
      "logParseExceptions" : false,
      "maxParseExceptions" : 0
    }
  },
  "hadoopDependencyCoordinates" : null,
  "classpathPrefix" : null,
  "context" : { }
}

Querying the basefact datasource provides the result as: "test_sketch":43672556.4879819 while the slice datasource results in "test_sketch":43676771.06402646

Still investigating the issue, but has anyone observed similar behavior?

[jihoonson]

I see some differences between two specs in the size of thetaSketch and partitionSpec. The thetaSketch gives an approximate result, and so I guess maybe the different result comes from the mismatch in size and the partitionSpec which is expected?

[himanshug]

from a cursory look it does look like the behavior due to sketch merge unpredictability and unrelated to #6743 .

[AlexanderSaydakov]

I am afraid that I don't know Druid enough to understand what does the above reindexing actually do with the sketches. Are they downsized in the process? Are the dimensional combinations altered in some way? Theta sketch algorithm is deterministic in a sense that merging the same sketches in different order still should produce exactly the same result.

[a2l007]

@AlexanderSaydakov Found out that this issue is not limited to reindexing. Indexing the source data with reduced dimensions also produces sketch values which are different from their equivalent sketches in the base datasource. Comparing and testing changes between 0.12.3 and 0.13.0, I could see that this issue started happening after #5335
@leventov Do you have any thoughts regarding this behaviour?

[somanullah]

@a2l007 @leventov We're seeing similar behavior on our cluster. It was recently upgraded to 0.14.0 and we have noticed differences in sketch results for the same data when comparing with data indexed from 0.12.1. Is there a fix for this?

[a2l007]

@somanullah What is the sketch size used in your case? I've managed to reproduce this issue internally but don't have a fix yet.
@leventov Could you help in giving some pointers on how to debug this issue, since this seems to entirely based on the rewrite of the index merging code via #5335. The overall gist of the issue is that the combined values of theta sketches after index merging using the RowPointer approach is different from the combined values using Rowboats merging. This mismatch happens only in case of a fairly large input data file. For example, the testcase I have currently has close to 110,000 events with a single thetasketch metric. Querying the indices merged using the RowBoat technique gives me the sketch value to be 1756 and querying those merged using the RowPointer technique gives the value as 1746. This difference widens as there are bigger indices to be merged.
I don't think this issue is related to sketch merge unpredictability since the sketch size and the data set are the same and this should return consistent results.

[leventov]

@a2l007 probably you could add debug printing in RowCombiningTimeAndDimsIterator to figure out which sketches combine into 1756 and 1746 and see what is missing.

[a2l007]

@leventov The issue is that when sketches are empty, the AggregateCombiner.reset() within SketchAggregatorFactory does not clear out the combined union object. As a result, this would have the combined value from a previous merge and this stale value would be used as the combined value for the present set of rows. I'm raising a PR shortly for fixing this.

[jihoonson]

@a2l007 thank you for digging into the problem!