<p dir="auto">I'm implementing a custom <a href="https://github.com/druid-io/druid/blob/master/processing/src/main/java/io/druid/query/aggregation/BufferAggregator.java">BufferAggregator</a>. The doc in the code level is pretty good! I just have extra questions that, as a dev user, could ask:</p>
<ul dir="auto">
<li>
<p dir="auto"><del>What does <a href="https://github.com/druid-io/druid/blob/master/processing/src/main/java/io/druid/query/aggregation/BufferAggregator.java#L54">buf</a> actually hold? Are they bytes of metrics or bytes of aggregators?</del></p>
</li>
<li>
<p dir="auto">When we say <a href="https://github.com/druid-io/druid/blob/master/processing/src/main/java/io/druid/query/aggregation/BufferAggregator.java#L45">"Implementations must not change the position, limit or mark of the given buffer"</a>, do we always want to use <code class="notranslate">buf.duplicate();</code> or <code class="notranslate">buf.slice()</code> as the first statement in implementation? Why does the position of the given buffer could possibly be non-zero? Is it because the same byteBuffer has been used in initialization of other kinds of aggregators? If so, when did the init happen?</p>
</li>
<li>
<p dir="auto">What's the difference between <code class="notranslate">Aggregator</code> and <code class="notranslate">BuffereAggregator</code>? Is <code class="notranslate">Aggregator</code> post agg and <code class="notranslate">BufferAggregator</code> non-post agg?</p>
</li>
</ul>
<h3 dir="auto">Affected Version</h3>
<p dir="auto">Druid Built From Source as of Fri Jun 5, 2020</p>
<h3 dir="auto">Description</h3>
<p dir="auto">We are trying to create new druid ingestion specs to pull from s3 directly via index_parallel rather than hadoop. The outputs simply don't seem to be partitioned though.</p>
<p dir="auto">To make it easy to reproduce, here is a simple config that shows the issue:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="{
  &quot;spec&quot;: {
    &quot;type&quot;: &quot;index_parallel&quot;,
    &quot;ioConfig&quot;: {
      &quot;type&quot;: &quot;index_parallel&quot;,
      &quot;inputSource&quot;: {
        &quot;type&quot;: &quot;http&quot;,
        &quot;uris&quot;: [
          &quot;https://druid.apache.org/data/wikipedia.json.gz&quot;,
          &quot;https://druid.apache.org/data/wikipedia.json.gz&quot;
        ]
      },
      &quot;inputFormat&quot;: {
        &quot;type&quot;: &quot;json&quot;
      }
    },
    &quot;tuningConfig&quot;: {
      &quot;type&quot;: &quot;index_parallel&quot;,
      &quot;partitionsSpec&quot;: {
        &quot;type&quot;: &quot;single_dim&quot;,
        &quot;partitionDimension&quot;: &quot;channel&quot;,
        &quot;maxRowsPerSegment&quot;: 2000
      },
      &quot;forceGuaranteedRollup&quot;: true,
      &quot;maxNumConcurrentSubTasks&quot;: 4
    },
    &quot;dataSchema&quot;: {
      &quot;dataSource&quot;: &quot;wikipedia-test-partitioned-2&quot;,
      &quot;granularitySpec&quot;: {
        &quot;type&quot;: &quot;uniform&quot;,
        &quot;segmentGranularity&quot;: &quot;DAY&quot;,
        &quot;queryGranularity&quot;: &quot;HOUR&quot;,
        &quot;rollup&quot;: true,
        &quot;intervals&quot;: [
          &quot;2000-01-01/2030-01-01&quot;
        ]
      },
      &quot;timestampSpec&quot;: {
        &quot;column&quot;: &quot;timestamp&quot;,
        &quot;format&quot;: &quot;iso&quot;
      },
      &quot;dimensionsSpec&quot;: {
        &quot;dimensions&quot;: [
          &quot;channel&quot;,
          &quot;cityName&quot;,
          &quot;comment&quot;,
          &quot;countryIsoCode&quot;,
          &quot;countryName&quot;,
          &quot;diffUrl&quot;,
          &quot;flags&quot;,
          &quot;isAnonymous&quot;,
          &quot;isMinor&quot;,
          &quot;isNew&quot;,
          &quot;isRobot&quot;,
          &quot;isUnpatrolled&quot;,
          &quot;metroCode&quot;,
          &quot;namespace&quot;,
          &quot;page&quot;,
          &quot;regionIsoCode&quot;,
          &quot;regionName&quot;,
          &quot;user&quot;
        ]
      },
      &quot;metricsSpec&quot;: [
        {
          &quot;name&quot;: &quot;count&quot;,
          &quot;type&quot;: &quot;count&quot;
        },
        {
          &quot;name&quot;: &quot;sum_added&quot;,
          &quot;type&quot;: &quot;longSum&quot;,
          &quot;fieldName&quot;: &quot;added&quot;
        },
        {
          &quot;name&quot;: &quot;sum_commentLength&quot;,
          &quot;type&quot;: &quot;longSum&quot;,
          &quot;fieldName&quot;: &quot;commentLength&quot;
        },
        {
          &quot;name&quot;: &quot;sum_deleted&quot;,
          &quot;type&quot;: &quot;longSum&quot;,
          &quot;fieldName&quot;: &quot;deleted&quot;
        },
        {
          &quot;name&quot;: &quot;sum_delta&quot;,
          &quot;type&quot;: &quot;longSum&quot;,
          &quot;fieldName&quot;: &quot;delta&quot;
        },
        {
          &quot;name&quot;: &quot;sum_deltaBucket&quot;,
          &quot;type&quot;: &quot;longSum&quot;,
          &quot;fieldName&quot;: &quot;deltaBucket&quot;
        }
      ]
    }
  },
  &quot;type&quot;: &quot;index_parallel&quot;
}"><pre class="notranslate"><code class="notranslate">{
  "spec": {
    "type": "index_parallel",
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "http",
        "uris": [
          "https://druid.apache.org/data/wikipedia.json.gz",
          "https://druid.apache.org/data/wikipedia.json.gz"
        ]
      },
      "inputFormat": {
        "type": "json"
      }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "partitionsSpec": {
        "type": "single_dim",
        "partitionDimension": "channel",
        "maxRowsPerSegment": 2000
      },
      "forceGuaranteedRollup": true,
      "maxNumConcurrentSubTasks": 4
    },
    "dataSchema": {
      "dataSource": "wikipedia-test-partitioned-2",
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "HOUR",
        "rollup": true,
        "intervals": [
          "2000-01-01/2030-01-01"
        ]
      },
      "timestampSpec": {
        "column": "timestamp",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": [
          "channel",
          "cityName",
          "comment",
          "countryIsoCode",
          "countryName",
          "diffUrl",
          "flags",
          "isAnonymous",
          "isMinor",
          "isNew",
          "isRobot",
          "isUnpatrolled",
          "metroCode",
          "namespace",
          "page",
          "regionIsoCode",
          "regionName",
          "user"
        ]
      },
      "metricsSpec": [
        {
          "name": "count",
          "type": "count"
        },
        {
          "name": "sum_added",
          "type": "longSum",
          "fieldName": "added"
        },
        {
          "name": "sum_commentLength",
          "type": "longSum",
          "fieldName": "commentLength"
        },
        {
          "name": "sum_deleted",
          "type": "longSum",
          "fieldName": "deleted"
        },
        {
          "name": "sum_delta",
          "type": "longSum",
          "fieldName": "delta"
        },
        {
          "name": "sum_deltaBucket",
          "type": "longSum",
          "fieldName": "deltaBucket"
        }
      ]
    }
  },
  "type": "index_parallel"
}
</code></pre></div>
<p dir="auto">Since maxRowsPerSegment is 2,000 and there are 24,000 rows in the dateset, I was expecting many partitions.</p>
<p dir="auto">I made sure to set two files so that it could be parallelized, since I saw some comments about needing to set maxNumConcurrentSubTasks and figured this might also apply to needing multiple input files.</p>
<p dir="auto">Of course the data that I have is much larger, coming out to a few GB, but I'm seeing the exact same issue.</p>
<p dir="auto">I have tried both setting targetRowsPerSegment and maxRowsPerSegment, and neither work.</p>