@Override public void write(Writable w) throws IOException {
    DruidWritable record = (DruidWritable) w;
    final long timestamp = (long) record.getValue().get(DruidConstants.DEFAULT_TIMESTAMP_COLUMN);
    final int
        partitionNumber =
        Math.toIntExact((long) record.getValue().getOrDefault(Constants.DRUID_SHARD_KEY_COL_NAME, -1L));
    final InputRow
        inputRow =
        new MapBasedInputRow(timestamp,
            dataSchema.getParser().getParseSpec().getDimensionsSpec().getDimensionNames(),
            record.getValue());

    try {

      if (partitionNumber != -1 && maxPartitionSize == -1) {
        /*
        Case data is sorted by time and an extra hashing dimension see DRUID_SHARD_KEY_COL_NAME
        Thus use DRUID_SHARD_KEY_COL_NAME as segment partition in addition to time dimension
        Data with the same DRUID_SHARD_KEY_COL_NAME and Time interval will end in the same segment
        */
        DateTime truncatedDateTime = segmentGranularity.bucketStart(DateTimes.utc(timestamp));
        final Interval interval = new Interval(truncatedDateTime, segmentGranularity.increment(truncatedDateTime));

        if (currentOpenSegment != null) {
          if (currentOpenSegment.getShardSpec().getPartitionNum() != partitionNumber
              || !currentOpenSegment.getInterval().equals(interval)) {
            pushSegments(ImmutableList.of(currentOpenSegment));
            currentOpenSegment =
                new SegmentIdWithShardSpec(dataSchema.getDataSource(),
                    interval,
                    tuningConfig.getVersioningPolicy().getVersion(interval),
                    new LinearShardSpec(partitionNumber));
          }
        } else {
          currentOpenSegment =
              new SegmentIdWithShardSpec(dataSchema.getDataSource(),
                  interval,
                  tuningConfig.getVersioningPolicy().getVersion(interval),
                  new LinearShardSpec(partitionNumber));

        }
        appenderator.add(currentOpenSegment, inputRow, committerSupplier::get);

      } else if (partitionNumber == -1 && maxPartitionSize != -1) {
        /*Case we are partitioning the segments based on time and max row per segment maxPartitionSize*/
        appenderator.add(getSegmentIdentifierAndMaybePush(timestamp), inputRow, committerSupplier::get);
      } else {
        throw new IllegalArgumentException(String.format(
            "partitionNumber and maxPartitionSize should be mutually exclusive "
                + "got partitionNum [%s] and maxPartitionSize [%s]",
            partitionNumber,
            maxPartitionSize));
      }

    } catch (SegmentNotWritableException e) {
      throw new IOException(e);
    }
  }
This function computes the segment identifier and push the current open segment
The push will occur if max size is reached or the event belongs to the next interval.
Note that this function assumes that timestamps are pseudo sorted.
This function will close and move to the next segment granularity as soon as
an event from the next interval appears. The sorting is done by the previous stage.

@return segmentIdentifier with of the truncatedTime and maybe push the current open segment.
druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java