I've been using PostgreSQL for so long I've forgotten how to spell SQL Server. My joiner tables are regular tables with the same base name as the primary table (business table "source.table_name_t1", digest table "source.table_name_h1", joiner table "source.table_name_j1"). They all share the primary key column(s), but t1 has the row values, h1 has the row digest, and j1 has the u_or_d column. I name the oid columns differently so the joined result is unambiguous.
INSERTs are executed immediately by an arcpy.da.InsertCursor. There really isn't likely to be much performance gain by writing to an extra table, then appending from that, though I suppose it depends on the number of indexes in the primary table. I have a further complication of needing to do this to a replica parent geodatabase, and using "INSERT INTO ... SELECT ..." with versioning would add unnecessary complexity. Performance-wise, the query from a source DBMS is the long pole in my tent, and SHA-1 hashing is somewhat expensive, especially when less than 1% of each table is modified, so there's large input I/O and compute costs, actually writing new rows is somewhat rare.
Generally speaking, I don't permit empty strings to kludge around NOT NULL constraints, but I'm not always able to enforce that. For purposes of hashing, I'd recommend treating an empty field in a CSV as equivalent to a NULL. I have a funky table or two where I have to backtrack to enforce NOT NULL on strings by replacing None with '' before INSERT, but doing it that way makes me feel dirty.
FWIW, I make extensive use of function dictionaries, so my encoder functions are placed in a dictionary by target datatype, and then they can massage oddities without harming the data stream. The code below handles empty string and NULL differently, (a single 0 byte vs a single ETX control character byte)
def pack_str(v,n):
if (v == None): return(b'\x00')
ba = bytearray()
if (type(v) == datetime.datetime):
if VERBOSE_ENABLED:
print(" {:8d}: Expected str, got {:s}!".format(n,str(type)))
v = v.strftime(isoFmt)
elif (type(v) != str):
if VERBOSE_ENABLED:
print(" {:8d}: Expected str, got {:s}!".format(n,str(type)))
v = str(v)
vt = v.rstrip(' ').encode('UTF-8')
ba.extend(vt)
ba.extend(b'\x03') # ASCII End-Of-Text (ETX)
return ba
def pack_i2(v,n):
if (v == None): return(b'\x00')
return struct.pack('h',v)
def pack_i4(v,n):
if (v == None): return(b'\x00')
return struct.pack('i',v)
...
packer_by_type = { # Lookup via native 'udt' value
'bool' : pack_i2,
'bpchar' : pack_str,
'bytea' : None, # BLOBs manifest as memoryview (unsed)
'char' : pack_str,
'date' : pack_date,
'float4' : pack_f8,
'float8' : pack_f8,
'geography' : pack_geom,
'geometry' : pack_geom,
'int2' : pack_i2,
'int4' : pack_i4,
'int8' : pack_i8,
'numeric' : pack_f8,
'timestamp' : pack_date,
'text' : pack_str,
'uuid' : pack_str,
'varchar' : pack_str,
}
# Note: When object is instantiated, the packing function is added
# to a "packers" list
...
def getHash(self,qRow):
self.rowCount += 1
ba1 = bytearray()
for i,packer in enumerate(self.packers,start=self.keyCount):
ba1 += packer(qRow[i],self.rowCount)
return hashlib.new('sha1',ba1,usedforsecurity=False).hexdigest()
