importccl: support default expressions

[pbardea]

It would be nice if IMPORT could support default column expressions. We probably want to plumb down an eval ctx and populate the defaults during row production.

[pbardea]

Expanding a bit on this here, I think in particular we want to support the following:

Given:
file.csv:

1,2
2,NULL

I think to start we should at least implement:

> IMPORT TABLE t1 (a INT, b INT DEFAULT 42) CSV DATA ('nodelocal://1/file.csv');
> SELECT * FROM t1;
  a |  b
----+-----
  1 |  2
  2 | 42
> CREATE TABLE t2 (a INT, b INT DEFAULT 42);
> IMPORT INTO t2 (a, b) CSV DATA ('nodelocal://1/file.csv');
> SELECT * FROM t2;
  a |  b
----+-----
  1 |  2
  2 | 42

I think it would also be reasonable to expect the following to work:

> CREATE TABLE t3 (a INT, b INT, c INT DEFAULT 42);
> IMPORT INTO t3 (a INT, b INT) CSV DATA ('nodelocal://1/file.csv');
> SELECT * FROM t3;
  a |  b   | c
----+------+-----
  1 |    2 | 42
  2 | NULL | 42

[Anzoteh96]

I wonder if replacing NULL in a targeted column with DEFAULT is the intended way for IMPORT, given that this doesn't seem like the case when I do INSERT INTO:

> CREATE TABLE t1 (a INT, b INT DEFAULT 75)
> INSERT INTO t1 (a, b) VALUES (33, NULL)
> SELECT * FROM t1
 a |  b
----+-----
33| NULL 

vs the following (where b is non-targeted column)

> CREATE TABLE t1 (a INT, b INT DEFAULT 75)
> INSERT INTO t1 (a) VALUES (33)
> SELECT * FROM t1
 a |  b
----+----
33|75

[pbardea]

Yep - good point. Default values can indeed be nullable and are only populated when not included in an insert statement.

[Anzoteh96]

Reopened again: the linked PR #50295 only handles constant default expressions: the next step will be to support other expressions with Stable volatility, like now() (those whose value do not change within the same statement, but may change across different statements). It might make sense to add support for unique_rowid() too.

[Anzoteh96]

Strategy for nextval
Goals / desired properties (in the order of priority, imho)

1. To ensure that the nextval values are unique, even after including those that have been populated by INSERT INTO.
2. To ensure that the nextval values are monotonically increasing (with respect to rows) within a file.
3. Ideally, the nextval values in a file should be consecutive (i.e. +1 between adjacent rows).

The last point, however, is difficult (if not impractical) to implement given that IMPORT including multiple files / tables can be done simultaneously. So it could be modified to make sure that there's minimal gaps between the numbers when they happen to be sorted (e.g. if 51 and 55 are both populated, we want 52, 53, 54 to be populated somewhere in the files too subject to the monotonicity constraint in (2)). Even so, the fact that inserts can be done during an IMPORT (into other tables that do not undergo the same IMPORT job but possibly use the same sequence(s) with the other tables being imported), makes this difficult (for the reasons described below). In addition, (2) is complicated by the fact that we have multiple processors processing the same import job simultaneously.

Strategy: from a table descriptor, the columns using any sequence and the corresponding sequences (under curval and nextval) can be extracted, from which we can determine at import planning stage (or perhaps, the beginning of the import stage itself) what are the sequences that we need to pay attention to. This means we can do the following:

1. For each sequence, the main processor / coordinator determines the curval (that is, either 0 or the value last used by nextval process).
2. Next, it supplies disjoint chunks of numbers for the processors (say there are 3 processors, then we could allocate numbers 1-1000, 1001-2000, 2001-3000 to processors 1, 2, 3, respectively). It also records / updates, internally, what is the last number being allocated (3000 in the example above). This has to be done in an atomic order to ensure that no other read and write (by, for example, other INSERTs of the same sequence(s) to other tables) to the sequence record is being done.
3. If a processor runs out of its slots, then it can request for more numbers (i.e. an additional chunk) on the fly from the coordinator, and the coordinator will simply update the last value allocated.

Notice that this is a modification from the previous strategy that says that at the end of an IMPORT, each processor communicates the highest number it used, and the highest of that +1 becomes the next value to be used for nextval. However, if an INSERT INTO (other table of same sequence) is used then it will be tricky to maintain the uniqueness of subsequent uses of this sequence. Consider the following scenario:

1. The chunks allocated are 1-1000, 1001-2000, 2001-3000 as per the example above.
2. An insert to other table using the same sequence happens in the middle of an IMPORT, and let's say it uses values 3001-3010.
3. The values last used by the 3 processors are 997, 1929, 2930.

The previous strategy would use 2931 as the next value for nextval but we might run into the problem of repeating the values 3001-3010 as per the insert (we could maintain a list / lookup to forbid 3001-3010 being used again but maintaining this list is more expensive compared to just maintaining a single number indicating the next value to be used for nextval).

Overall, this strategy should be safe to adopt as it maintains the uniqueness property; a drawback is that it could potentially leave a huge gap between numbers (say, we allocate 1000 numbers but only 10 rows are being imported. That's a gap of 990 numbers).

---

Separate from this: once this is supported, curval should also be supported too since we're basically reading the value off to know what's the most recent value being used from the lookup map. One scenario that could make this tricky is when we have nextval() and curval() simultaneously:

CREATE SEQUENCE myseq // fresh sequence, so starts at 0
CREATE TABLE t (a INT, b INT DEFAULT nextval('myseq'), c INT DEFAULT curval('myseq')
IMPORT INTO t (blah)

In this case: which curval are we using? The value right before IMPORT starts (in this case, 0), or the most updated value of myseq value after updated via nextval? The latter could be tricky because imagine we have something as crazy as

nextval('myseq') + curval('myseq') * nextval('myseq')

as default expression. Then it's not even clear what should be the value of curval.

[dt]

This sounds about right ,but I don't know if we need the main proc / coordinator in the picture at all -- when we run out of IDs in a given processor, we'll need to reserve more, and, as you say, that reservation needs to be done atomically from the point of view of any other traffic to the sequence unrelated to the IMPORT so we'll use a txn. If we have a txn, we don't really need to go back to the coordinator to coordinate, since if multiple processors need a chunk at the same time, the txn should serialize them on its own, and that avoids a lot of mess trying to do 2-way data flow / rpcs between processors and the coordinator.

The other thing I'd do is start with smaller chunk size -- e.g. reserve 10 initially, then on subsequent reservations go back for an order of magnitude more each time until we hit, say, 100k. That way small tables of 5 values don't blow a 100k hole in the sequence, but big tables don't spend all their time contending on the sequence reservations.

[maddyblue]

If you have a txn during an IMPORT won't that block any INSERTs from happening at the same time? Is that ok? Also, unless it's changed, sequence increments happen in their own internal txn such that they are atomic, but not transactional (with respect to any user-created transaction). That is, an aborted transaction that had an INSERT would still result in an incremented sequence.

[maddyblue]

Oh, it's for the above reason (atomic but not transactional sequence increments) that I'm not sure why you need that property during IMPORT. Users already can't have transactional sequences and INSERT ids be created out-of-order (for example, multiple INSERT transactions can finish in some order and have a different order for their sequence IDs). Why do IMPORTs need that property?

[dt]

@mjibson I'm not sure I follow. What property?

The plan is that when an IMPORT processor hits a row that needs a next-val, it'll go to an txn to atomically a) update of the sequence to take a chunk of IDs and b) record the file and line number at which that chunk was taken in the job progress. Then on resume, if we previously reserved a chunk for a given line, we'll use it. That is an important property of IMPORT, in that it ensures every time you re-run the same IMPORT (i.e. resume the same job), you produce the same KVs, so if you resume after a pause or crash, your uncheckpointed prior work is perfectly shadowed by your re-done version. If you abort the IMPORT, the "holes" are still in the sequence keyspace -- we updated it in a short-lived txn that committed on the first resume, regardless of what happens later.

[maddyblue]

I was referring to 3 in the previous comment:

3. Ideally, the nextval values in a file should be consecutive (i.e. +1 between adjacent rows).

[dt]

Ah, we sort of want that property since if used in keys, we'd like adjacent input rows to be adjacent KVs for ingestion performance. And simple incrementing within our reserved chunk is also just the easiest way to know when we've exhausted the chunk and know we need to get another one. But we don't make any effort to ensure sequential chunks of the same file have sequential IDs.

[Anzoteh96]

Following the consideration of "determinism" (that is, on resume we produce the same KVs), an additional thing we'll need to do is to include both the value to be used next and the chunk limit (together with the resume position) when saving progress.

[Anzoteh96]

Here's one tricky scenario that I can think of:

Imagine we saved at row 30 (say), last used value 1024, chunk limit 1100.

1. Let's also assume that after we save that, we use up more slots and reach our limit, ending up asking for more. Let's also assume that the next chunk is 1401-1500.
2. If the processor had to resume (after a failure, say), we'll resume at 1024, reach the chunk limit 1100. But now, it will have to remember that 1401-1500 belongs to the processor itself. Should we consider caching the chunks that a processor has asked for?

Separate from this, I concur with the idea of starting with a smaller chunk size and increasing this chunk size as we progress.

[pbardea]

But now, it will have to remember that 1401-1500 belongs to the processor itself.

We shouldn't need to remember which processor requested which chunk. All we need to know is which file and line number corresponds to which chunk. These (file num, row num, reserved-chunk) entries should be encoded in the import job details in the same txn that reserves the chunk. On resume, the processor should lookup if there is already a chunk reserved from the previous attempt.

[mwang1026]

Closing this. We didn't get to next_val(sequence) so I'll create separate tracking for this. Awesome stuff @Anzoteh96 !