Ability to auto populate timestamp on event ingestion.

[bhrgv-bolla]

Managed timestamp. Always incremental.

[manishmalhotrawork]

@bhrgv-bolla Can you please add more details to it?
Are we saying, add System.nano/milli to each event if user says use ingestion time or user has not defined a timestamp column?

[bhrgv-bolla]

Yes, timestamp on ingestion. I made a hack for our use case.

This is a slightly modified string input row parser.

/*
 * Licensed to Metamarkets Group Inc. (Metamarkets) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. Metamarkets 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 com.example.druid.parsers;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Iterators;
import com.google.common.collect.Maps;
import io.druid.data.input.ByteBufferInputRowParser;
import io.druid.data.input.InputRow;
import io.druid.data.input.impl.MapInputRowParser;
import io.druid.data.input.impl.ParseSpec;
import io.druid.java.util.common.collect.Utils;
import io.druid.java.util.common.parsers.ParseException;
import io.druid.java.util.common.parsers.Parser;
import org.joda.time.DateTime;

import javax.annotation.Nullable;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CoderResult;
import java.nio.charset.CodingErrorAction;
import java.util.List;
import java.util.Map;

/**
 * Adds a timestamp column to the current time.
 * User needs to specify timestamp column in timestamp spec.
 */
public class ManagedStringInputRowParser implements ByteBufferInputRowParser {
    private static final Charset DEFAULT_CHARSET = Charsets.UTF_8;

    private final ParseSpec parseSpec;
    private final MapInputRowParser mapParser;
    private final Charset charset;

    private Parser<String, Object> parser;
    private CharBuffer chars;
    private final Boolean autoTimeStamp;

    @JsonProperty
    public String getType() {
        return "managed";
    }

    @JsonCreator
    public ManagedStringInputRowParser(
            @JsonProperty("parseSpec") ParseSpec parseSpec,
            @JsonProperty("encoding") String encoding,
            @JsonProperty("autoTimestamp") Boolean autoTimeStamp
    ) {
        this.parseSpec = Preconditions.checkNotNull(parseSpec, "parseSpec");
        this.mapParser = new MapInputRowParser(parseSpec);

        //Enabled by default.
        if(autoTimeStamp == null) this.autoTimeStamp = true;
        else this.autoTimeStamp = autoTimeStamp;

        if (encoding != null) {
            this.charset = Charset.forName(encoding);
        } else {
            this.charset = DEFAULT_CHARSET;
        }
    }

    @Deprecated
    public ManagedStringInputRowParser(ParseSpec parseSpec) {
        this(parseSpec, null, false);
    }

    public List<InputRow> parseBatch(ByteBuffer input) {
        return Utils.nullableListOf(parseMap(buildStringKeyMap(input)));
    }

    @JsonProperty
    public ParseSpec getParseSpec() {
        return parseSpec;
    }

    @JsonProperty
    public String getEncoding() {
        return charset.name();
    }

    public ManagedStringInputRowParser withParseSpec(ParseSpec parseSpec) {
        return new ManagedStringInputRowParser(parseSpec, getEncoding(), true);
    }

    private Map<String, Object> buildStringKeyMap(ByteBuffer input) {
        int payloadSize = input.remaining();

        if (chars == null || chars.remaining() < payloadSize) {
            chars = CharBuffer.allocate(payloadSize);
        }

        final CoderResult coderResult = charset.newDecoder()
                .onMalformedInput(CodingErrorAction.REPLACE)
                .onUnmappableCharacter(CodingErrorAction.REPLACE)
                .decode(input, chars, true);

        Map<String, Object> theMap;
        if (coderResult.isUnderflow()) {
            chars.flip();
            try {
                theMap = parseString(chars.toString());
                theMap = Maps.newHashMap(theMap); //ObjectFlatnner causing issues.
                //Always a timestamp column need to exist
                if(getParseSpec() == null || getParseSpec().getTimestampSpec() == null || Strings.isNullOrEmpty(getParseSpec().getTimestampSpec().getTimestampColumn())) {
                    throw new IllegalStateException("Couldn't find the timestamp column name : "+ getParseSpec());
                }
                String timestampColumn = getParseSpec().getTimestampSpec().getTimestampColumn();
                if(theMap != null && this.autoTimeStamp) theMap.put(timestampColumn, DateTime.now().toDateTimeISO().toString()); //TODO review Managed timestamp.
            } finally {
                chars.clear();
            }
        } else {
            throw new ParseException("Failed with CoderResult[%s]", coderResult);
        }
        return theMap;
    }

    public void initializeParser() {
        if (parser == null) {
            // parser should be created when it is really used to avoid unnecessary initialization of the underlying
            // parseSpec.
            parser = parseSpec.makeParser();
        }
    }

    public void startFileFromBeginning() {
        initializeParser();
        parser.startFileFromBeginning();
    }

    @Nullable
    public InputRow parse(@Nullable String input) {
        return parseMap(parseString(input));
    }

    @Nullable
    private Map<String, Object> parseString(@Nullable String inputString) {
        initializeParser();
        return parser.parseToMap(inputString);
    }

    @Nullable
    private InputRow parseMap(@Nullable Map<String, Object> theMap) {
        // If a header is present in the data (and with proper configurations), a null is returned
        if (theMap == null) {
            return null;
        }
        return Iterators.getOnlyElement(mapParser.parseBatch(theMap).iterator());
    }
}

[kaijianding]

Don't think it is a good idea, it is not idempotent. if a realtime node restarts, the data timestamp becomes different. @bhrgv-bolla

[gianm]

@kaijianding makes a good point - this strategy won't play well with ingest from Kafka / Kinesis (on replay, the timestamp will change). It might be better to support Kafka's native timestamp field. Currently we don't. There's an issue open for it: #6625

[stale]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the dev@druid.apache.org list. Thank you for your contributions.

[stale]

This issue has been closed due to lack of activity. If you think that is incorrect, or the issue requires additional review, you can revive the issue at any time.

[benkrug]

What's the risk of the operation not being idempotent? If a realtime node restarts and it gets a new ingestion timestamp, does that cause some further issues? Wouldn't only one value be persisted and published? (Or, if not, could that just be a known risk?) Or, if you have to restart with useEarliestOffset, you might get a different value, but couldn't that be a risk that could be known, like duplicate values is a risk? This seems to be a common ask.

[nttoan26]

let the developer makes decision on idempotent, there are many cases this feature is helpful where idempotent is not important

[sergioferragut]

Hi, I'm very new to druid, I think this subject is important in at least the IoT space and even though this issue is old, ingestion timestamp calculation continues to come up as an issue/request/question.

I think there are different data pipeline timestamps involved when streaming event message:

• the event timestamp : I think this is what most in this discussion are concerned with when this time value is used as the primary partitioning timestamp. When the timestamp value is in the streamed message, then it is idempotent, if we use a system time it would not be. Using system time for this doesn't make sense to me regardless of whether idempotent or not, if we use system time at ingestion for this, then the time of the event is meaningless unless the event we are modeling IS the ingestion of the message...which I guess could be a use case.
• published timestamp when the message was published to the message bus, this is the timestamp on the message envelope - it is useful to measure lag between the event and publication; it does not seem to make sense to use the druid system time at ingestion for this at all because it just isn't representative of the publish time, if there are performance issues or recoveries, it could potentially be a long time between publication and ingestion.
• ingestion timestamp when the rows are ingested into the database. The immediate use of this is to be able to measure event to ingestion lag. This is what I believe most comments on the subject are referring to. Isn't this essentially the ability to use a now() or current_timestamp() function in a transformSpec expression? I think the more relevant case is when this value is ingested as a Dimension and not as the primary timestamp or maybe as a lag metric calculated as (now() - event_time)

Anyway, I think it comes down to adding now() or current_timestamp() that return system time in transformation and metric specifications at ingestion time.

Sorry for the long comment. :-)

[nibinqtl]

being able to measure the lag is important for my use case. a now() function will be very helpful!

[gianm]

Unfortunately, there's still this issue with determinism. It's important that two servers reading the same stream will evaluate expressions the same way, which means we can't have expressions like now() or random() that are evaluated on a per-row basis. As to letting the user make the decision -- imo, we should avoid features like that, because even though they can be useful, they are foot-guns that make it easy to accidentally mess up your state.

We did recently add #11630, which makes it possible to use the Kafka-assigned timestamp when reading from Kafka streams. I think that would help in a lot of these cases -- not all -- but hopefully a good amount. It's also totally safe, which is a plus!